130 lines
3.8 KiB
Python
130 lines
3.8 KiB
Python
|
|
"""Test greeter failure/retry/launch against a fake greetd server.
|
||
|
|
|
||
|
|
Usage: QML2_IMPORT_PATH=<qt5compat>/lib/qt-6/qml python3 scripts/greeter-check.py <auth-bundle>
|
||
|
|
Requires qs on PATH. No real authentication or desktop launch occurs.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import json, os, socket, struct, subprocess as sp, tempfile, time, shutil, sys
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
qs = shutil.which("qs")
|
||
|
|
assert qs, "qs must be on PATH"
|
||
|
|
root = Path(tempfile.mkdtemp(prefix="fred-greetd-check."))
|
||
|
|
root.chmod(0o700)
|
||
|
|
shutil.copytree(str(Path(sys.argv[1]) / "greeter"), root / "config")
|
||
|
|
p = root / "config/shell.qml"
|
||
|
|
p.chmod(0o600)
|
||
|
|
s = (
|
||
|
|
p.read_text()
|
||
|
|
.replace("import Quickshell\n", "import Quickshell\nimport Quickshell.Io\n")
|
||
|
|
.replace(" AuthView {", " AuthView {\n id: auth")
|
||
|
|
)
|
||
|
|
s = (
|
||
|
|
s.rstrip()[:-1]
|
||
|
|
+ """ IpcHandler { target: "test"; function submit(password: string): void { auth.submitted(password); } }
|
||
|
|
}
|
||
|
|
"""
|
||
|
|
)
|
||
|
|
p.write_text(s)
|
||
|
|
server = socket.socket(socket.AF_UNIX)
|
||
|
|
server.bind(str(root / "greetd.sock"))
|
||
|
|
server.listen()
|
||
|
|
server.settimeout(10)
|
||
|
|
env = {
|
||
|
|
**os.environ,
|
||
|
|
"HOME": str(root),
|
||
|
|
"XDG_RUNTIME_DIR": str(root),
|
||
|
|
"GREETD_SOCK": str(root / "greetd.sock"),
|
||
|
|
"FRED_GREETER": "1",
|
||
|
|
"QT_QPA_PLATFORM": "offscreen",
|
||
|
|
"QT_QUICK_BACKEND": "software",
|
||
|
|
"FRED_SESSION_COMMAND": "/fixture/authenticated-session",
|
||
|
|
}
|
||
|
|
log = open(root / "log", "w")
|
||
|
|
proc = sp.Popen([qs, "-p", str(root / "config")], env=env, stdout=log, stderr=log)
|
||
|
|
try:
|
||
|
|
conn, _ = server.accept()
|
||
|
|
conn.settimeout(5)
|
||
|
|
|
||
|
|
def recv():
|
||
|
|
data = b""
|
||
|
|
while len(data) < 4:
|
||
|
|
data += conn.recv(4 - len(data))
|
||
|
|
size = struct.unpack("<I", data)[0]
|
||
|
|
data = b""
|
||
|
|
while len(data) < size:
|
||
|
|
data += conn.recv(size - len(data))
|
||
|
|
value = json.loads(data)
|
||
|
|
print(value, flush=True)
|
||
|
|
return value
|
||
|
|
|
||
|
|
def send(value):
|
||
|
|
data = json.dumps(value).encode()
|
||
|
|
conn.sendall(struct.pack("<I", len(data)) + data)
|
||
|
|
|
||
|
|
def submit():
|
||
|
|
for _ in range(30):
|
||
|
|
result = sp.run(
|
||
|
|
[
|
||
|
|
qs,
|
||
|
|
"ipc",
|
||
|
|
"--pid",
|
||
|
|
str(proc.pid),
|
||
|
|
"call",
|
||
|
|
"test",
|
||
|
|
"submit",
|
||
|
|
"fixture-password",
|
||
|
|
],
|
||
|
|
env=env,
|
||
|
|
capture_output=True,
|
||
|
|
text=True,
|
||
|
|
)
|
||
|
|
if result.returncode == 0:
|
||
|
|
return
|
||
|
|
time.sleep(0.1)
|
||
|
|
raise AssertionError(result.stderr)
|
||
|
|
|
||
|
|
for success in [False, True]:
|
||
|
|
submit()
|
||
|
|
assert recv() == {"type": "create_session", "username": "fred"}
|
||
|
|
send(
|
||
|
|
{
|
||
|
|
"type": "auth_message",
|
||
|
|
"auth_message_type": "secret",
|
||
|
|
"auth_message": "Password:",
|
||
|
|
}
|
||
|
|
)
|
||
|
|
assert recv() == {
|
||
|
|
"type": "post_auth_message_response",
|
||
|
|
"response": "fixture-password",
|
||
|
|
}
|
||
|
|
if success:
|
||
|
|
send({"type": "success"})
|
||
|
|
launch = recv()
|
||
|
|
assert (
|
||
|
|
launch["type"] == "start_session"
|
||
|
|
and launch["cmd"] == ["/fixture/authenticated-session"]
|
||
|
|
and "XDG_SESSION_TYPE=wayland" in launch["env"]
|
||
|
|
)
|
||
|
|
send({"type": "success"})
|
||
|
|
assert proc.wait(timeout=5) == 0
|
||
|
|
else:
|
||
|
|
send(
|
||
|
|
{
|
||
|
|
"type": "error",
|
||
|
|
"error_type": "auth_error",
|
||
|
|
"description": "fixture rejection",
|
||
|
|
}
|
||
|
|
)
|
||
|
|
assert recv() == {"type": "cancel_session"}
|
||
|
|
send({"type": "success"})
|
||
|
|
time.sleep(0.3)
|
||
|
|
print(
|
||
|
|
"Greeter: failed password does not launch; retry and authenticated launch pass"
|
||
|
|
)
|
||
|
|
finally:
|
||
|
|
proc.terminate()
|
||
|
|
proc.wait()
|
||
|
|
log.close()
|
||
|
|
print((root / "log").read_text())
|