295 lines
15 KiB
Python
295 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
"""Synthetic wizard/Keychain tests. Run with the openconnect-lite Python."""
|
|
import base64
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import fcntl
|
|
from pathlib import Path
|
|
import pty
|
|
import select
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import termios
|
|
import time
|
|
|
|
sys.dont_write_bytecode = True
|
|
|
|
# The existing tests/logging.py is a test entry point, not stdlib logging.
|
|
sys.path = [p for p in sys.path if Path(p).resolve() != Path(__file__).resolve().parent]
|
|
from unittest.mock import patch
|
|
|
|
import pyotp
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
spec = importlib.util.spec_from_file_location("yandex_import", ROOT / "bin/lemana-vpn-import-yandex.py")
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
SECRET = base64.b32encode(b"12345678901234567890").decode()
|
|
OTHER = "JBSWY3DPEHPK3PXP"
|
|
URI = f"otpauth://totp/Lemana:example?secret={SECRET}&algorithm=SHA1&digits=6&period=30"
|
|
ENTRY = {"name": "Lemana VPN", "secret": SECRET, "techInfo": URI}
|
|
ORIGINAL_CONFIG = 'LEMANA_VPN_USERNAME="example"\nLEMANA_VPN_MODE="bitwarden-auto"\nLEMANA_VPN_LOCAL_OTP="0"\nLEMANA_VPN_LOG_MAX_BYTES="12345"\n'
|
|
verify = pyotp.TOTP.verify
|
|
assert pyotp.TOTP(SECRET).at(59) == "287082" # RFC 6238 SHA1 vector, six digits.
|
|
|
|
|
|
class FakeKeyring:
|
|
def __init__(self, previous=OTHER):
|
|
self.values = {"example": "unchanged-password"}
|
|
if previous is not None:
|
|
self.values["totp/example"] = previous
|
|
self.writes = []
|
|
|
|
def get_password(self, service, account):
|
|
assert service == "openconnect-lite"
|
|
return self.values.get(account)
|
|
|
|
def set_password(self, service, account, value):
|
|
assert service == "openconnect-lite" and account == "totp/example"
|
|
self.writes.append((account, value))
|
|
self.values[account] = value
|
|
|
|
def delete_password(self, service, account):
|
|
assert service == "openconnect-lite" and account == "totp/example"
|
|
self.values.pop(account, None)
|
|
|
|
|
|
def wizard(directory, keyring, answers=None, entries=None, code="287082"):
|
|
backup = directory / "key_backup"
|
|
backup.write_text(json.dumps(entries if entries is not None else [
|
|
{"name": "Yandex account", "techInfo": "otpauth://yaotp/test?secret=not-supported"}, ENTRY]))
|
|
config = directory / "env"
|
|
config.write_text(ORIGINAL_CONFIG)
|
|
inputs = iter(answers if answers is not None else [str(backup), "2"])
|
|
output = []
|
|
|
|
def ask(prompt):
|
|
output.append(prompt)
|
|
try:
|
|
return next(inputs)
|
|
except StopIteration:
|
|
raise EOFError from None
|
|
|
|
def ask_secret(prompt):
|
|
assert any("LDAP: example" in line for line in output), output
|
|
assert any("Запись: Lemana VPN" in line for line in output), output
|
|
output.append(prompt)
|
|
if code is None:
|
|
raise EOFError
|
|
return code
|
|
|
|
failure = None
|
|
with patch.dict(sys.modules, {"keyring": keyring}), patch.object(pyotp.TOTP, "verify", lambda self, otp, valid_window: verify(self, otp, for_time=59, valid_window=valid_window)):
|
|
try:
|
|
module.configure("example", config, ask, ask_secret, output.append)
|
|
except (module.ImportFailure, EOFError, KeyboardInterrupt, OSError) as error:
|
|
failure = error
|
|
captured = "\n".join(output)
|
|
assert SECRET not in captured and OTHER not in captured and URI not in captured, captured
|
|
assert "unchanged-password" not in captured
|
|
return config, failure, output
|
|
|
|
|
|
with tempfile.TemporaryDirectory(prefix="lemana-yandex-") as temporary:
|
|
directory = Path(temporary)
|
|
keyring = FakeKeyring()
|
|
config, error, output = wizard(directory, keyring)
|
|
assert error is None, error
|
|
assert keyring.writes == [("totp/example", SECRET)], keyring.writes
|
|
assert keyring.values["example"] == "unchanged-password"
|
|
assert 'LEMANA_VPN_MODE="keychain-auto"' in config.read_text()
|
|
assert 'LEMANA_VPN_LOCAL_OTP="1"' in config.read_text()
|
|
assert 'LEMANA_VPN_LOG_MAX_BYTES="12345"' in config.read_text()
|
|
assert config.stat().st_mode & 0o777 == 0o600
|
|
assert not list(directory.glob(".otp-env-*"))
|
|
print("PASS wizard: mixed backup, explicit VPN selection, RFC code, one Keychain entry and private settings")
|
|
|
|
for answers, code in (([], "287082"), ([str(directory / "key_backup")], "287082"),
|
|
([str(directory / "key_backup"), ""], "287082"),
|
|
([str(directory / "key_backup"), "0"], "287082"),
|
|
(None, "000000"), (None, None)):
|
|
keyring = FakeKeyring()
|
|
config, error, _ = wizard(directory, keyring, answers=answers, code=code)
|
|
assert error is not None and keyring.writes == []
|
|
assert config.read_text() == ORIGINAL_CONFIG
|
|
print("PASS cancellation/invalid choice/wrong code: no writes")
|
|
|
|
variants = [URI.replace("//totp/", "//hotp/"), URI.replace("//totp/", "//yaotp/"),
|
|
URI.replace("SHA1", "SHA256"), URI.replace("digits=6", "digits=8"),
|
|
URI.replace("period=30", "period=60"), URI + "&secret=" + OTHER,
|
|
URI + "&algorithm=SHA1", URI + "&digits=6", URI + "&period=30",
|
|
URI.replace(SECRET, "NOT-BASE32"), URI.replace(SECRET, OTHER),
|
|
URI + "&counter=1", URI.replace("otpauth:", "https:")]
|
|
variants.extend([URI + "&encoder=steam", URI + "&chars=alpha&length=8"])
|
|
for uri in variants:
|
|
keyring = FakeKeyring()
|
|
config, error, _ = wizard(directory, keyring, entries=[dict(ENTRY, techInfo=uri)],
|
|
answers=[str(directory / "key_backup"), "1"])
|
|
assert isinstance(error, module.ImportFailure), (uri, error)
|
|
assert keyring.writes == [] and config.read_text() == ORIGINAL_CONFIG
|
|
assert SECRET not in str(error) and OTHER not in str(error)
|
|
print("PASS unsupported types/parameters, duplicate parameters, malformed/mismatched secrets")
|
|
|
|
for previous in (OTHER, None):
|
|
for failure in (OSError("secret backend detail"), KeyboardInterrupt()):
|
|
keyring = FakeKeyring(previous)
|
|
with patch.object(module.os, "replace", side_effect=failure):
|
|
config, error, _ = wizard(directory, keyring)
|
|
assert error is failure and config.read_text() == ORIGINAL_CONFIG
|
|
assert keyring.get_password("openconnect-lite", "totp/example") == previous
|
|
assert not list(directory.glob(".otp-env-*"))
|
|
keyring = FakeKeyring()
|
|
with patch.object(keyring, "set_password", side_effect=RuntimeError("backend secret")), patch.dict(sys.modules, {"keyring": keyring}):
|
|
try:
|
|
module.activate(SECRET, "example", config)
|
|
assert False, "expected backend failure"
|
|
except module.ImportFailure as error:
|
|
assert "вернуть прежний секрет не удалось" in str(error) and "backend secret" not in str(error)
|
|
assert config.read_text() == ORIGINAL_CONFIG
|
|
print("PASS config failure/interruption rollback and secret-safe rollback failure")
|
|
|
|
keyring = FakeKeyring()
|
|
real_replace = module.os.replace
|
|
|
|
def interrupted_after_replace(source, destination):
|
|
real_replace(source, destination)
|
|
raise KeyboardInterrupt
|
|
|
|
with patch.object(module.os, "replace", side_effect=interrupted_after_replace):
|
|
config, error, _ = wizard(directory, keyring)
|
|
assert isinstance(error, module.ImportFailure) and "уже сохранена" in str(error)
|
|
assert keyring.values["totp/example"] == SECRET and 'LEMANA_VPN_LOCAL_OTP="1"' in config.read_text()
|
|
keyring = FakeKeyring(None)
|
|
with patch.object(keyring, "set_password", return_value=None):
|
|
config, error, _ = wizard(directory, keyring)
|
|
assert isinstance(error, module.ImportFailure) and config.read_text() == ORIGINAL_CONFIG
|
|
keyring = FakeKeyring()
|
|
config.write_text(ORIGINAL_CONFIG.replace('USERNAME="example"', 'USERNAME="another"'))
|
|
with patch.dict(sys.modules, {"keyring": keyring}):
|
|
try:
|
|
module.activate(SECRET, "example", config)
|
|
assert False, "wrong saved account accepted"
|
|
except module.ImportFailure:
|
|
pass
|
|
assert not keyring.writes
|
|
config.write_text(ORIGINAL_CONFIG)
|
|
print("PASS commit-point consistency, silent backend failure and saved-account binding")
|
|
|
|
backup = directory / "bad_backup"
|
|
for data in (b"not JSON", b"{}", b"[]", b"[{\"techInfo\":1}]", b"x" * (module.MAX_BACKUP_BYTES + 1)):
|
|
backup.write_bytes(data)
|
|
try:
|
|
module.read_backup(backup)
|
|
assert False, "bad backup accepted"
|
|
except module.ImportFailure:
|
|
pass
|
|
assert module.display_name("hello\x1b[2J\nworld") == "hello[2Jworld"
|
|
assert module.display_name("A" * 110 + SECRET, {SECRET}) == ""
|
|
keyring = FakeKeyring()
|
|
config, error, output = wizard(directory, keyring, entries=[{"name": SECRET.lower(), "techInfo": URI}],
|
|
answers=[str(directory / "key_backup"), ""])
|
|
assert SECRET.lower() not in "\n".join(output) and not keyring.writes
|
|
config, error, output = wizard(directory, keyring, entries=[{
|
|
"name": SECRET, "secret": " " + SECRET + " ", "techInfo": URI.replace(SECRET, "%20" + SECRET + "%20")}],
|
|
answers=[str(directory / "key_backup"), ""])
|
|
assert SECRET not in "\n".join(output) and not keyring.writes
|
|
print("PASS malformed/oversized files and terminal-control sanitization")
|
|
|
|
# Installed entry point must fail without a controlling terminal, before writes.
|
|
sso = directory / "xdg/openconnect-lite"
|
|
sso.mkdir(parents=True)
|
|
sso.joinpath("config.toml").write_text('[credentials]\nusername="example"\n')
|
|
env = dict(os.environ, HOME=temporary, XDG_CONFIG_HOME=str(sso.parent), XDG_CONFIG_DIRS=str(sso.parent))
|
|
proc = subprocess.run([sys.executable, str(ROOT / "bin/lemana-vpn-import-yandex.py"),
|
|
"--username=example", "--config", str(config)],
|
|
stdin=subprocess.DEVNULL, capture_output=True, text=True, start_new_session=True, env=env)
|
|
assert proc.returncode == 1 and "Traceback" not in proc.stderr
|
|
assert config.read_text() == ORIGINAL_CONFIG
|
|
print("PASS no-terminal entry: bounded error, no writes")
|
|
|
|
# Drive the actual CLI + helper through an automated PTY and a fake backend.
|
|
fake = directory / "fake"
|
|
fake.mkdir()
|
|
fake.joinpath("security").write_text("#!/bin/sh\nexit 0\n")
|
|
fake.joinpath("security").chmod(0o755)
|
|
state = directory / "keyring.json"
|
|
state.write_text(json.dumps({"example": "unchanged-password"}))
|
|
fake.joinpath("keyring.py").write_text('''import json, os
|
|
from pathlib import Path
|
|
state = Path(os.environ["TEST_KEYRING"])
|
|
def get_password(service, account):
|
|
assert service == "openconnect-lite"
|
|
return json.loads(state.read_text()).get(account)
|
|
def set_password(service, account, password):
|
|
assert service == "openconnect-lite"
|
|
data = json.loads(state.read_text()); data[account] = password
|
|
state.write_text(json.dumps(data))
|
|
def delete_password(service, account):
|
|
data = json.loads(state.read_text()); data.pop(account, None)
|
|
state.write_text(json.dumps(data))
|
|
''')
|
|
directory.joinpath("key_backup").write_text(json.dumps([ENTRY]))
|
|
env = {key: value for key, value in env.items() if not key.startswith(("LEMANA_VPN_", "OPENCONNECT_LITE_"))}
|
|
env.update(PATH=str(fake) + ":" + os.environ["PATH"], PYTHONPATH=str(fake), TEST_KEYRING=str(state),
|
|
LEMANA_VPN_CONFIG_DIR=temporary, LEMANA_VPN_OC_PYTHON=sys.executable,
|
|
OPENCONNECT_LITE_CONFIG_DIR=str(sso), LEMANA_VPN_STATUS_DIR=str(directory / "status"))
|
|
master, slave = pty.openpty()
|
|
|
|
def controlling_terminal():
|
|
os.setsid()
|
|
fcntl.ioctl(slave, termios.TIOCSCTTY, 0)
|
|
|
|
proc = subprocess.Popen(["/bin/bash", str(ROOT / "bin/vpn-lemanapro.sh"), "--import-yandex-key"],
|
|
stdin=slave, stdout=slave, stderr=slave, env=env, preexec_fn=controlling_terminal)
|
|
os.close(slave)
|
|
output = b""
|
|
phone_code = pyotp.TOTP(SECRET).now()
|
|
inputs = [("Путь к файлу", str(directory / "key_backup")), ("Номер записи", "1"),
|
|
("для проверки и сохранения:", phone_code)]
|
|
deadline = time.monotonic() + 15
|
|
try:
|
|
while time.monotonic() < deadline:
|
|
if select.select([master], [], [], max(0, deadline - time.monotonic()))[0]:
|
|
try:
|
|
chunk = os.read(master, 8192)
|
|
except OSError:
|
|
break # PTY EOF on macOS.
|
|
if not chunk:
|
|
break
|
|
output += chunk
|
|
if inputs and inputs[0][0].encode() in output:
|
|
os.write(master, (inputs.pop(0)[1] + "\n").encode())
|
|
if proc.poll() is not None:
|
|
break
|
|
assert proc.wait(timeout=2) == 0, output.decode(errors="replace")
|
|
finally:
|
|
os.close(master)
|
|
if proc.poll() is None:
|
|
proc.kill()
|
|
proc.wait()
|
|
assert not inputs and SECRET.encode() not in output and phone_code.encode() not in output
|
|
assert json.loads(state.read_text()) == {"example": "unchanged-password", "totp/example": SECRET}
|
|
status = subprocess.run(["bash", str(ROOT / "bin/vpn-lemanapro.sh"), "--status", "--json"],
|
|
env=env, text=True, capture_output=True, check=True)
|
|
modules = json.loads(status.stdout)["modules"]
|
|
assert modules["local_otp"] and modules["mode"]["ready"] and not modules["bitwarden"]["enabled"]
|
|
override = subprocess.run(["bash", str(ROOT / "bin/vpn-lemanapro.sh"), "--status", "--json"],
|
|
env=dict(env, LEMANA_VPN_MODE="manual-keychain", LEMANA_VPN_LOCAL_OTP="0"), text=True, capture_output=True, check=True)
|
|
assert not json.loads(override.stdout)["modules"]["local_otp"]
|
|
config_after = config.read_text()
|
|
stored_after = state.read_text()
|
|
runtime_definitions = (ROOT / "bin/vpn-lemanapro.sh").read_text().split('\nif $STATUS_MODE; then', 1)[0]
|
|
mismatch = subprocess.run(["bash", "-c", runtime_definitions + '\n_can_prompt() { return 0; }; _import_yandex_key'],
|
|
env=dict(env, LEMANA_VPN_USERNAME="different-user"), text=True, capture_output=True)
|
|
assert mismatch.returncode != 0 and "отличается" in mismatch.stderr
|
|
mismatch = subprocess.run([sys.executable, str(ROOT / "bin/lemana-vpn-import-yandex.py"),
|
|
"--username=different-user", "--config", str(config)],
|
|
env=env, text=True, capture_output=True)
|
|
assert mismatch.returncode != 0 and "профиле SSO" in mismatch.stderr
|
|
assert config.read_text() == config_after and state.read_text() == stored_after
|
|
print("PASS real CLI/PTY import → fake Keychain → persisted mode → status; hidden code; override and account guards")
|
|
|
|
print("yandex import checks ok")
|