#!/usr/bin/env python3 """End-to-end OTP login flow against a real PocketBase: request -> verify -> select-identity. Covers the select-identity step the original suite missed (reading the grant's json identity_ids via getStringSlice). Needs KQ_POCKETBASE_BIN. A local HTTP server stands in for Evolution and captures the code that is sent. """ import json import os import re import shutil import socket import subprocess import sys import tempfile import threading import time import unittest import urllib.error import urllib.request from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path HERE = Path(__file__).resolve().parent SCRIPTS = HERE.parent.parent / "scripts" def free_port(): with socket.socket() as s: s.bind(("127.0.0.1", 0)) return s.getsockname()[1] class EvolutionHandler(BaseHTTPRequestHandler): def do_POST(self): length = int(self.headers.get("Content-Length", "0")) body = json.loads(self.rfile.read(length) or b"{}") with self.server.lock: self.server.messages.append(body) self.send_response(200) self.send_header("Content-Type", "application/json") self.end_headers() self.wfile.write(b'{"key":{"id":"x"}}') def log_message(self, *_): return class OtpSelectFlowTest(unittest.TestCase): @classmethod def setUpClass(cls): binary = os.environ.get("KQ_POCKETBASE_BIN", "") if not binary or not Path(binary).is_file(): raise unittest.SkipTest("KQ_POCKETBASE_BIN is required") cls.temp = tempfile.TemporaryDirectory(prefix="kq-otp-flow-") root = Path(cls.temp.name) cls.data_dir = root / "pb_data" cls.hooks_dir = root / "pb_hooks" cls.hooks_dir.mkdir(parents=True) for f in ("main.pb.js", "auth-service.js", "auth-utils.js"): shutil.copy2(HERE / f, cls.hooks_dir / f) cls.evo = ThreadingHTTPServer(("127.0.0.1", free_port()), EvolutionHandler) cls.evo.messages = [] cls.evo.lock = threading.Lock() threading.Thread(target=cls.evo.serve_forever, daemon=True).start() cls.port = free_port() cls.base = f"http://127.0.0.1:{cls.port}" cls.admin_email = "otp-admin@kq.local" cls.admin_password = "otp-only-password-123" subprocess.run([binary, "superuser", "upsert", cls.admin_email, cls.admin_password, "--dir", str(cls.data_dir), "--hooksDir", str(cls.hooks_dir)], check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) env = os.environ.copy() env.update({ "EVOLUTION_URL": f"http://127.0.0.1:{cls.evo.server_port}", "EVOLUTION_INSTANCE": "Kings-and-Queens", "EVOLUTION_APIKEY": "test-key", "KQ_OTP_PEPPER": "test-pepper", }) cls.log = tempfile.TemporaryFile(mode="w+") cls.pb = subprocess.Popen([binary, "serve", "--http", f"127.0.0.1:{cls.port}", "--dir", str(cls.data_dir), "--hooksDir", str(cls.hooks_dir), "--hooksWatch=false", "--dev=false"], env=env, stdout=cls.log, stderr=subprocess.STDOUT) cls._wait() _, auth = cls.req("POST", "/api/collections/_superusers/auth-with-password", {"identity": cls.admin_email, "password": cls.admin_password}) cls.token = auth["token"] cls._prereqs() cls._schema(binary) cls._student() @classmethod def tearDownClass(cls): for attr, fn in (("pb", lambda x: (x.terminate(), x.wait(timeout=5))), ("evo", lambda x: (x.shutdown(), x.server_close())), ("log", lambda x: x.close()), ("temp", lambda x: x.cleanup())): obj = getattr(cls, attr, None) if obj is not None: try: fn(obj) except Exception: pass @classmethod def _wait(cls): deadline = time.time() + 20 while time.time() < deadline: if cls.pb.poll() is not None: cls.log.seek(0) raise RuntimeError(cls.log.read()) try: with urllib.request.urlopen(f"{cls.base}/api/health", timeout=0.5): return except (urllib.error.URLError, TimeoutError): time.sleep(0.1) raise RuntimeError("pb did not start") @classmethod def req(cls, method, path, payload=None, token=None): body = json.dumps(payload).encode() if payload is not None else None r = urllib.request.Request(cls.base + path, data=body, method=method) r.add_header("Content-Type", "application/json") if token: r.add_header("Authorization", token) try: with urllib.request.urlopen(r, timeout=10) as resp: raw = resp.read().decode() return resp.status, (json.loads(raw) if raw else {}) except urllib.error.HTTPError as e: raw = e.read().decode() return e.code, (json.loads(raw) if raw else {}) @classmethod def _prereqs(cls): for d in ({"name": "classes", "type": "base", "fields": [ {"name": "date", "type": "text", "required": True}, {"name": "student_id", "type": "text"}, {"name": "status", "type": "select", "maxSelect": 1, "values": ["scheduled", "cancelled"]}]}, {"name": "student_profiles", "type": "base", "fields": [ {"name": "user_id", "type": "text", "required": True}, {"name": "phone", "type": "text"}, {"name": "is_active", "type": "bool"}]}): s, b = cls.req("POST", "/api/collections", d, cls.token) if s not in (200, 201): raise RuntimeError(f"prereq {d['name']} failed: {b}") @classmethod def _schema(cls, binary): env = os.environ.copy() env.update({"KQ_PB_URL": cls.base, "KQ_PB_ADMIN_EMAIL": cls.admin_email, "KQ_PB_ADMIN_PASSWORD": cls.admin_password}) r = subprocess.run([sys.executable, str(SCRIPTS / "setup_auth_family.py")], env=env, capture_output=True, text=True) if r.returncode != 0: raise RuntimeError(f"schema failed: {r.stdout}\n{r.stderr}") @classmethod def _student(cls): cls.phone = "5492915550000" s, u = cls.req("POST", "/api/collections/users/records", { "email": "alumno@kq.local", "password": "studentpass123", "passwordConfirm": "studentpass123", "rol": "student", "name": "Alumno Test", "phone": cls.phone, "otp_enabled": True, "verified": True, "emailVisibility": True, }, cls.token) if s not in (200, 201): raise RuntimeError(f"student create failed: {u}") cls.student_id = u["id"] s, p = cls.req("POST", "/api/collections/student_profiles/records", { "user_id": cls.student_id, "phone": cls.phone, "is_active": True, }, cls.token) if s not in (200, 201): raise RuntimeError(f"profile create failed: {p}") def _code_from_evolution(self): deadline = time.time() + 5 while time.time() < deadline: with self.evo.lock: msgs = list(self.evo.messages) for m in reversed(msgs): match = re.search(r"(\d{6})", m.get("text", "")) if match: return match.group(1) time.sleep(0.1) raise AssertionError("Evolution never received an OTP code") def test_full_login_flow_reaches_token(self): # 1. request-otp -> generic 200, code dispatched to (mock) Evolution. s, b = self.req("POST", "/api/kq/auth/request-otp", {"phone": "2915550000"}) self.assertEqual(s, 200, b) self.assertEqual(b, {"ok": True}) code = self._code_from_evolution() # 2. verify-otp -> grant + the student identity. s, b = self.req("POST", "/api/kq/auth/verify-otp", {"phone": "2915550000", "code": code}) self.assertEqual(s, 200, b) self.assertIn("grant", b) ids = [i["id"] for i in b.get("identities", [])] self.assertIn(self.student_id, ids, b) # 3. select-identity -> a real auth response with token + record (the bug). s, b = self.req("POST", "/api/kq/auth/select-identity", {"grant": b["grant"], "user_id": self.student_id}) self.assertEqual(s, 200, b) self.assertIn("token", b) self.assertTrue(b.get("token")) self.assertEqual(b.get("record", {}).get("id"), self.student_id, b) # 4. the grant is now single-use. s2, _ = self.req("POST", "/api/kq/auth/select-identity", {"grant": "x" * 64, "user_id": self.student_id}) self.assertEqual(s2, 400) if __name__ == "__main__": unittest.main(verbosity=2)