170 lines
9.0 KiB
Python
170 lines
9.0 KiB
Python
#!/usr/bin/env python3
|
|
"""DNS command boundary with fake sudo/helper; never touches system DNS."""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
CLI = ROOT / "bin/vpn-lemanapro.sh"
|
|
|
|
with tempfile.TemporaryDirectory(prefix="lemana-dns-cli-") as temporary:
|
|
home = Path(temporary)
|
|
fake = home / "bin"
|
|
fake.mkdir()
|
|
helper = home / "DNS helper's script"
|
|
for name in ("security", "openconnect", "bw", "networksetup"):
|
|
(fake / name).write_text('#!/bin/sh\necho forbidden >> "$HOME/forbidden"\nexit 1\n')
|
|
# Status may check Keychain presence, but repair must not even read credentials.
|
|
(fake / "security").write_text('#!/bin/sh\necho keychain >> "$HOME/keychain-read"\nexit 1\n')
|
|
(fake / "sudo").write_text('''#!/bin/sh
|
|
[ "$1" = -n ] || exit 90
|
|
shift
|
|
printf '%s\n' "$@" > "$HOME/sudo-args"
|
|
[ "${TEST_SUDO_FAIL:-0}" = 0 ] || exit 1
|
|
exec "$@"
|
|
''')
|
|
for path in fake.iterdir():
|
|
path.chmod(0o755)
|
|
env = {k: v for k, v in os.environ.items() if not k.startswith(("LEMANA_VPN_", "OPENCONNECT_LITE_"))}
|
|
env.update(HOME=temporary, PATH=str(fake) + ":" + os.environ["PATH"],
|
|
LEMANA_VPN_CONFIG_DIR=str(home / "config"), LEMANA_VPN_STATUS_DIR=str(home / "status"),
|
|
LEMANA_VPN_OC_VENV=str(home / "venv"), LEMANA_VPN_OC_BIN=str(fake / "openconnect"),
|
|
LEMANA_VPN_DNS_CLEANUP=str(helper))
|
|
|
|
def repair(**extra):
|
|
result = subprocess.run(["bash", str(CLI), "--repair-dns", "--json"],
|
|
env=dict(env, **extra), text=True, capture_output=True, timeout=10)
|
|
payload = json.loads(result.stdout)
|
|
assert payload["event"] == "dns_recovery"
|
|
assert not (home / "forbidden").exists()
|
|
assert not (home / "keychain-read").exists()
|
|
assert not (home / "status").exists()
|
|
return result.returncode, payload["result"]
|
|
|
|
assert repair() == (1, "error")
|
|
helper.write_text('#!/bin/sh\necho old-helper-ran >> "$HOME/forbidden"\n')
|
|
helper.chmod(0o755)
|
|
assert repair() == (1, "error")
|
|
assert not (home / "sudo-args").exists(), "Legacy helper must not be executed"
|
|
helper.write_text('''#!/bin/sh
|
|
# LEMANA_DNS_PROTOCOL=2
|
|
[ "$#" = 1 ] && [ "$1" = --recover ] || exit 91
|
|
printf '%s\n' "${TEST_DNS_RESULT:-unchanged}"
|
|
exit "${TEST_DNS_EXIT:-0}"
|
|
''')
|
|
for value in ("restored", "unchanged", "no_snapshot", "active"):
|
|
assert repair(TEST_DNS_RESULT=value) == (0, value)
|
|
assert (home / "sudo-args").read_text().splitlines() == [str(helper), "--recover"]
|
|
for extra in ({"TEST_DNS_RESULT": "not-json\nwrong"}, {"TEST_DNS_EXIT": "1"}, {"TEST_SUDO_FAIL": "1"}):
|
|
assert repair(**extra) == (1, "error")
|
|
|
|
source = CLI.read_text()
|
|
def function(name):
|
|
start = source.index(name + "() {")
|
|
return source[start:source.index("\n}\n", start) + 3]
|
|
|
|
cleanup = "\n".join(function(name) for name in ("_emit", "_dns_recovery_supported", "_dns_cleanup"))
|
|
for client_exit, recovery_exit in ((0, 0), (7, 0), (7, 1)):
|
|
ended = subprocess.run(["bash", "-c", cleanup + '\nJSON_MODE=true\n_dns_cleanup "$TEST_CLIENT_EXIT"'],
|
|
env=dict(env, DNS_CLEANUP=str(helper), TEST_CLIENT_EXIT=str(client_exit),
|
|
TEST_DNS_EXIT=str(recovery_exit)), text=True, capture_output=True, timeout=10)
|
|
assert ended.returncode == client_exit, ended.stderr
|
|
events = [json.loads(line) for line in ended.stdout.splitlines()]
|
|
assert events[0]["event"] == "dns_cleanup"
|
|
assert events[-1] == {"event": "dns_recovery", "result": "error" if recovery_exit else "unchanged"}
|
|
|
|
helper.write_text('''#!/bin/sh
|
|
# LEMANA_DNS_PROTOCOL=2
|
|
count=0
|
|
[ ! -f "$HOME/recovery-count" ] || read -r count < "$HOME/recovery-count"
|
|
count=$((count + 1))
|
|
printf '%s\\n' "$count" > "$HOME/recovery-count"
|
|
if [ "$count" -le "$TEST_ACTIVE_COUNT" ]; then
|
|
printf 'active\\n'
|
|
else
|
|
touch "$HOME/dns-restored"
|
|
printf 'restored\\n'
|
|
fi
|
|
''')
|
|
(fake / "sleep").write_text('#!/bin/sh\nexit 0\n')
|
|
(fake / "sleep").chmod(0o755)
|
|
for active_count in (3, 100):
|
|
(home / "recovery-count").unlink(missing_ok=True)
|
|
(home / "dns-restored").unlink(missing_ok=True)
|
|
ended = subprocess.run(["bash", "-c", cleanup + '\nJSON_MODE=true\n_dns_cleanup 130'],
|
|
env=dict(env, DNS_CLEANUP=str(helper), TEST_ACTIVE_COUNT=str(active_count)),
|
|
text=True, capture_output=True, timeout=10)
|
|
assert ended.returncode == 130, ended.stderr
|
|
events = [json.loads(line) for line in ended.stdout.splitlines()]
|
|
assert events[-1] == {"event": "dns_recovery", "result": "restored" if active_count == 3 else "active"}
|
|
assert (home / "dns-restored").exists() == (active_count == 3)
|
|
assert int((home / "recovery-count").read_text()) == (4 if active_count == 3 else 30)
|
|
print("PASS disconnect recovery: waits for exiting VPN children, preserves exit code, bounds active-session retries")
|
|
|
|
status = subprocess.run(["bash", str(CLI), "--status", "--json"], env=env,
|
|
text=True, capture_output=True, check=True, timeout=10)
|
|
assert json.loads(status.stdout)["modules"]["dns_cleanup"]["recovery_supported"]
|
|
assert not (home / "forbidden").exists()
|
|
# Exercise the actual OpenConnect shell command with a path containing quotes.
|
|
hook_line = next(line for line in CLI.read_text().splitlines() if line.startswith("dns_hook="))
|
|
helper.write_text('#!/bin/sh\nprintf "%s\\n" "$@" > "$HOME/hook-args"\n')
|
|
subprocess.run(["bash", "-c", hook_line + '\n/bin/sh -c "$dns_hook"'],
|
|
env=dict(env, DNS_CLEANUP=str(helper), SESSION_ID='11111111-2222-3333-4444-555555555555'), check=True, timeout=10)
|
|
assert (home / "hook-args").read_text().splitlines() == ["--vpnc", "11111111-2222-3333-4444-555555555555"]
|
|
print("PASS DNS CLI: noninteractive recovery, fixed JSON, no connection/credentials, legacy capability guard")
|
|
|
|
(home / 'keychain-read').unlink(missing_ok=True)
|
|
(home / 'sudo-args').unlink(missing_ok=True)
|
|
token = '11111111-2222-3333-4444-555555555555'
|
|
def stop(**extra):
|
|
result = subprocess.run(['bash', str(CLI), '--stop-session=' + token, '--json'],
|
|
env=dict(env, **extra), text=True, capture_output=True, timeout=10)
|
|
assert not (home/'keychain-read').exists() and not (home/'forbidden').exists()
|
|
assert not (home/'status').exists()
|
|
return result.returncode, json.loads(result.stdout)
|
|
assert stop() == (1, {'event': 'session_stop', 'result': 'error'})
|
|
assert not (home/'sudo-args').exists(), 'An old helper must never receive --stop'
|
|
helper.write_text('''#!/bin/sh
|
|
# LEMANA_SESSION_PROTOCOL=1
|
|
[ "$#" = 2 ] && [ "$1" = --stop ] || exit 91
|
|
printf '%s\\n' "${TEST_STOP_RESULT:-stopping}"
|
|
exit "${TEST_STOP_EXIT:-0}"
|
|
''')
|
|
assert stop() == (0, {'event': 'session_stop', 'result': 'stopping'})
|
|
assert (home/'sudo-args').read_text().splitlines() == [str(helper), '--stop', token]
|
|
for extra in ({'TEST_STOP_RESULT': 'wrong'}, {'TEST_STOP_EXIT': '7'}, {'TEST_SUDO_FAIL': '1'}):
|
|
assert stop(**extra) == (1, {'event': 'session_stop', 'result': 'error'})
|
|
for invalid in ('', '../../etc/passwd', token + ';echo bad'):
|
|
result = subprocess.run(['bash', str(CLI), '--stop-session=' + invalid, '--json'], env=env,
|
|
text=True, capture_output=True, timeout=10)
|
|
assert result.returncode == 2
|
|
print('PASS stop CLI: exact session forwarded, no credentials/connect, unsupported helper and failures rejected')
|
|
|
|
attempt = '22222222-2222-3333-4444-555555555555'
|
|
restart_args = ['--restart-session='+token, '--restart-attempt='+attempt]
|
|
def restart(args=restart_args, **extra):
|
|
result = subprocess.run(['bash', str(CLI), *args], env=dict(env, **extra),
|
|
text=True, capture_output=True, timeout=10)
|
|
assert not (home/'keychain-read').exists() and not (home/'forbidden').exists()
|
|
assert not (home/'status').exists()
|
|
return result
|
|
(home/'sudo-args').unlink()
|
|
assert restart().returncode != 0 and not (home/'sudo-args').exists()
|
|
helper.write_text('''#!/bin/sh
|
|
# LEMANA_SESSION_PROTOCOL=1
|
|
# LEMANA_RECONNECT_PROTOCOL=1
|
|
[ "$#" = 3 ] && [ "$1" = --restart ] || exit 91
|
|
printf '%s\\n' "${TEST_RESTART_RESULT:-stopping}"
|
|
''')
|
|
for value in ('stopping','stale','cancelled'):
|
|
result = restart(TEST_RESTART_RESULT=value)
|
|
assert result.returncode == 0 and result.stdout.strip() == value
|
|
assert (home/'sudo-args').read_text().splitlines() == [str(helper), '--restart', token, attempt]
|
|
assert restart(TEST_RESTART_RESULT='unexpected').returncode != 0
|
|
assert restart(TEST_SUDO_FAIL='1').returncode != 0
|
|
for args in ([restart_args[0]], [restart_args[1]], [restart_args[0], '--restart-attempt=invalid']):
|
|
assert restart(args).returncode == 2
|
|
print('PASS restart CLI: exact session/attempt pair, capability gate, missing/invalid pair and helper failure rejected')
|