tools/pair_check.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""Verify device pairing against a real panel.
An API token is forty-odd characters. Nobody types that into a phone, which is
why a native client has been impractical to set up. Pairing trades a short
code, chosen and displayed by the box, for a real token.
What this holds down, in the order it matters:
1. A correct code returns a token, and that token actually works.
2. The code works exactly once.
3. A wrong code is refused, and no run of wrong guesses from a stranger can
lock out the real one (the 30**8 space, not a counter, stops guessing).
4. Minting a code needs authentication. Only redeeming one is open, and it is
open because the device doing it has no credential yet.
5. Every refusal reads the same. Distinguishing wrong from expired from none
outstanding tells a guesser where they are.
6. Cancelling takes a code back off the screen.
Its own home, port and tmux socket; nothing touches a panel you are using.
python3 tools/pair_check.py
Exit status is 0 on pass, 1 on fail.
"""
from __future__ import annotations
import json
import os
import shutil
import subprocess
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
PASSWORD = "pair-check" # noqa: S105 — a throwaway panel on loopback
PORT = 3289
BASE = f"http://127.0.0.1:{PORT}"
HOME = Path("/tmp/clique-pair-check-home")
SOCKET = "clique-pair-check"
def _panel() -> tuple[subprocess.Popen, str]:
shutil.rmtree(HOME, ignore_errors=True)
HOME.mkdir(parents=True)
env = dict(os.environ, CLIQUE_HOME=str(HOME), CLIQUE_TMUX_SOCKET=SOCKET)
subprocess.run(
[sys.executable, "-m", "clique", "password"],
input=f"{PASSWORD}\n{PASSWORD}\n",
text=True, env=env, cwd=str(ROOT), capture_output=True,
)
mint = subprocess.run(
[sys.executable, "-m", "clique", "token", "create", "admin"],
env=env, cwd=str(ROOT), capture_output=True, text=True,
).stdout
token = next((ln.strip() for ln in mint.splitlines() if ln.strip().startswith("mxp_")), "")
proc = subprocess.Popen(
[sys.executable, "-m", "clique", "--host", "127.0.0.1", "--port", str(PORT),
"--state", str(HOME / "state.json")],
env=env, cwd=str(ROOT), stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
)
for _ in range(80):
try:
urllib.request.urlopen(BASE + "/healthz", timeout=2).read()
return proc, token
except (urllib.error.URLError, OSError):
time.sleep(0.25)
proc.kill()
raise SystemExit(f"the check's own panel never came up on {PORT}")
def main() -> int:
proc, admin = _panel()
def call(path, method="GET", body=None, token=None):
headers = {"Content-Type": "application/json"}
if token:
headers["Authorization"] = "Bearer " + token
req = urllib.request.Request(
BASE + path,
data=(json.dumps(body).encode() if body is not None else None),
method=method, headers=headers,
)
try:
with urllib.request.urlopen(req, timeout=15) as r:
return json.loads(r.read() or "{}"), r.status
except urllib.error.HTTPError as err:
return json.loads(err.read() or "{}"), err.code
res: dict[str, object] = {}
try:
# Minting is not open. Only redeeming is.
_, status = call("/api/pair", "POST")
res["minting_a_code_needs_authentication"] = status == 401
started, status = call("/api/pair", "POST", token=admin)
code = str(started.get("code", ""))
res["an_authenticated_caller_gets_a_code"] = status == 200 and len(code) == 9
res["and_it_is_shown_in_two_groups"] = code[4:5] == "-"
res["with_no_characters_people_mistype"] = not (set(code) & set("01IOLU"))
# A phone types it with the dash and in whatever case it feels like.
paired, status = call(
"/api/pair/claim", "POST", {"code": code.lower(), "name": "Justin's Pixel"}
)
handed = str(paired.get("token", ""))
res["a_correct_code_hands_over_a_token"] = status == 201 and handed.startswith("mxp_")
res["and_the_token_is_named_after_the_device"] = paired.get("name") == "Justin's Pixel"
# The whole point: it has to actually work.
state, status = call("/api/state", token=handed)
res["and_that_token_really_works"] = status == 200 and "sessions" in state
again, status = call("/api/pair/claim", "POST", {"code": code})
res["a_code_cannot_be_used_twice"] = status == 403
# A wrong code, and then a run of wrong codes that must not lock out
# the live one: redeeming is open, so a counter would be a stranger's
# off switch.
fresh = str(call("/api/pair", "POST", token=admin)[0].get("code", ""))
wrong, wrong_status = call("/api/pair/claim", "POST", {"code": "ZZZZ-ZZZZ"})
res["a_wrong_code_is_refused"] = wrong_status == 403
res["and_says_nothing_about_why"] = wrong.get("error") == again.get("error")
for _ in range(15):
call("/api/pair/claim", "POST", {"code": "ZZZZ-ZZZZ"})
_, status = call("/api/pair/claim", "POST", {"code": fresh})
res["wrong_guesses_cannot_lock_out_the_real_code"] = status == 201
# Cancelling takes it back off the screen.
third = str(call("/api/pair", "POST", token=admin)[0].get("code", ""))
res["a_live_code_reports_its_time_left"] = call("/api/pair", token=admin)[0][
"expires_in"
] > 100
call("/api/pair", "DELETE", token=admin)
res["cancelling_clears_it"] = call("/api/pair", token=admin)[0]["expires_in"] == 0
_, status = call("/api/pair/claim", "POST", {"code": third})
res["and_a_cancelled_code_no_longer_works"] = status == 403
# Browser sign-in: the same code, posted as a form field, issues a
# session cookie. The Android JSON claim above must keep working; this
# is the other redemption, not a replacement.
browser_code = str(call("/api/pair", "POST", token=admin)[0].get("code", ""))
form = urllib.parse.urlencode({"pair": browser_code}).encode()
req = urllib.request.Request(
BASE + "/", data=form, method="POST",
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
try:
with urllib.request.urlopen(req, timeout=15) as r:
signed = r.status, r.headers.get("Set-Cookie", ""), r.read()
except urllib.error.HTTPError as err:
signed = err.code, err.headers.get("Set-Cookie", ""), err.read()
status, cookie, body = signed
res["a_code_signs_a_browser_in"] = (
status == 200 and cookie.startswith("clique=") and b"Signed in" in body
)
again = urllib.request.Request(
BASE + "/",
data=urllib.parse.urlencode({"pair": browser_code}).encode(),
method="POST",
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
try:
with urllib.request.urlopen(again, timeout=15) as r:
reuse_status, reuse_body = r.status, r.read()
except urllib.error.HTTPError as err:
reuse_status, reuse_body = err.code, err.read()
res["and_not_a_second_time"] = (
reuse_status == 401 and b"That code is no longer valid." in reuse_body
)
missing = urllib.request.Request(
BASE + "/",
data=urllib.parse.urlencode({"pair": "ZZZZ-ZZZZ"}).encode(),
method="POST",
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
try:
with urllib.request.urlopen(missing, timeout=15) as r:
miss_status, miss_body = r.status, r.read()
except urllib.error.HTTPError as err:
miss_status, miss_body = err.code, err.read()
res["a_wrong_code_is_the_same_no"] = (
miss_status == 401 and b"That code is no longer valid." in miss_body
)
page_req = urllib.request.Request(BASE + "/?pair=" + browser_code)
with urllib.request.urlopen(page_req, timeout=15) as r:
page = r.read()
res["GET_pair_serves_the_sign_in_page"] = (
b'name="pair"' in page and b"Sign in on this device" in page
)
finally:
proc.terminate()
subprocess.run(["tmux", "-L", SOCKET, "kill-server"],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
ok = all(bool(v) for v in res.values())
for key, value in res.items():
print(f" {'ok ' if value else 'FAIL'} {key}: {value}")
print("pair_check:", "PASS" if ok else "FAIL")
return 0 if ok else 1
if __name__ == "__main__":
raise SystemExit(main())