175 lines
8.7 KiB
Python
175 lines
8.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Process-level Laravel browser migration checks using real pinned PHP fixtures.
|
|
|
|
Usage: python3 tests/laravel_session_http.py /path/to/twofauth-rs
|
|
All databases, users, cookies and session files are disposable synthetic fixtures.
|
|
Regenerate fixtures with tests/fixtures/generate_laravel_sessions.php.
|
|
"""
|
|
import base64
|
|
from http.cookies import SimpleCookie
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import socket
|
|
import sqlite3
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
NATIVE_COOKIE = "rust_2fauth_session"
|
|
|
|
|
|
class Browser:
|
|
def __init__(self, base, cookies=None):
|
|
self.base = base
|
|
self.cookies = dict(cookies or {})
|
|
|
|
def request(self, path, method="GET", csrf=None, update=True):
|
|
headers = {"Accept": "application/json"}
|
|
if self.cookies:
|
|
headers["Cookie"] = "; ".join(f"{k}={v}" for k, v in self.cookies.items())
|
|
if csrf is not None:
|
|
headers["X-XSRF-TOKEN"] = csrf
|
|
request = urllib.request.Request(self.base + path, headers=headers, method=method)
|
|
try:
|
|
response = urllib.request.urlopen(request, timeout=10)
|
|
except urllib.error.HTTPError as error:
|
|
response = error
|
|
with response:
|
|
raw = response.read()
|
|
try:
|
|
body = json.loads(raw)
|
|
except (ValueError, UnicodeDecodeError):
|
|
body = raw.decode()
|
|
if update:
|
|
for header in response.headers.get_all("Set-Cookie", []):
|
|
parsed = SimpleCookie()
|
|
parsed.load(header)
|
|
for name, value in parsed.items():
|
|
if value["max-age"] == "0":
|
|
self.cookies.pop(name, None)
|
|
else:
|
|
self.cookies[name] = value.value
|
|
return response.status, body, response.headers
|
|
|
|
|
|
def run_case(binary, fixture, driver):
|
|
label = f"{driver}/{fixture['serialization']}/encrypt={fixture['encrypted']}"
|
|
with tempfile.TemporaryDirectory(prefix="laravel-browser-") as folder:
|
|
work = Path(folder)
|
|
storage = work / "storage"
|
|
sessions = storage / "framework/sessions"
|
|
sessions.mkdir(parents=True)
|
|
env = dict(os.environ, DATABASE_URL=f"sqlite://{work}/test.sqlite",
|
|
APP_KEY="base64:" + base64.b64encode(fixture["app_key"].encode()).decode(),
|
|
APP_NAME="2FAuth", SESSION_COOKIE="2fauth_session", SESSION_DRIVER=driver,
|
|
SESSION_ENCRYPT=str(fixture["encrypted"]).lower(), SESSION_TABLE="sessions",
|
|
SESSION_SECURE_COOKIE="false", SESSION_PATH="/", SESSION_DOMAIN="",
|
|
STORAGE_PATH=str(storage), RUST_COOKIE_SECURE="false",
|
|
RUST_PUBLIC_DIR=str(ROOT / "public"))
|
|
for name in ("APP_SUBDIRECTORY", "APP_PREVIOUS_KEYS", "AUTH_GUARD", "AUTHENTICATION_GUARD"):
|
|
env.pop(name, None)
|
|
subprocess.run([str(binary), "init"], env=env, cwd=work, check=True, capture_output=True)
|
|
db = sqlite3.connect(work / "test.sqlite")
|
|
db.execute("INSERT INTO users(id,name,email,password,remember_token) VALUES(7,'PHP fixture','fixture@example.invalid','unused','remember-fixture')")
|
|
db.execute("CREATE TABLE IF NOT EXISTS sessions(id TEXT PRIMARY KEY,user_id INTEGER,ip_address TEXT,user_agent TEXT,payload TEXT,last_activity INTEGER)")
|
|
sid = fixture["session_id"]
|
|
source = sessions / sid
|
|
if driver == "file":
|
|
source.write_text(fixture["payload"])
|
|
else:
|
|
db.execute("INSERT INTO sessions(id,user_id,payload,last_activity) VALUES(?,7,?,?)",
|
|
(sid, fixture["database_payload"], int(time.time())))
|
|
db.commit()
|
|
with socket.socket() as reserve:
|
|
reserve.bind(("127.0.0.1", 0))
|
|
port = reserve.getsockname()[1]
|
|
base = f"http://127.0.0.1:{port}"
|
|
process = subprocess.Popen([str(binary), "serve", "--listen", f"127.0.0.1:{port}"],
|
|
env=env, cwd=work, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
try:
|
|
anonymous = Browser(base)
|
|
for _ in range(100):
|
|
if process.poll() is not None:
|
|
raise AssertionError(f"{label}: server exited during startup")
|
|
try:
|
|
if anonymous.request("/up")[0] == 200:
|
|
break
|
|
except urllib.error.URLError:
|
|
time.sleep(.05)
|
|
else:
|
|
raise AssertionError(f"{label}: startup timed out")
|
|
original = {"2fauth_session": fixture["session_cookie"],
|
|
"XSRF-TOKEN": fixture["csrf_cookie"]}
|
|
|
|
# A browser need not visit the SPA before refreshing its Passport cookie.
|
|
direct = Browser(base, original)
|
|
assert direct.request("/oauth/token/refresh", "POST")[0] == 419, label
|
|
status, body, _ = direct.request("/oauth/token/refresh", "POST", fixture["csrf_cookie"])
|
|
assert status == 200 and body == "Refreshed.", (label, status, body)
|
|
assert "laravel_token" in direct.cookies, label
|
|
status, user, _ = direct.request("/api/v1/user", csrf=fixture["csrf_cookie"])
|
|
assert status == 200 and user["id"] == 7, (label, status, user)
|
|
|
|
# Start another browser with exactly the PHP session + recaller: no
|
|
# laravel_token and no native-session cookie exist before root GET.
|
|
original[fixture["recaller_name"]] = fixture["recaller_cookie"]
|
|
browser = Browser(base, original)
|
|
assert "laravel_token" not in browser.cookies and NATIVE_COOKIE not in browser.cookies
|
|
status, html, _ = browser.request("/")
|
|
assert status == 200 and "var appSettings" in html, (label, status)
|
|
assert NATIVE_COOKIE in browser.cookies, label
|
|
assert browser.cookies["XSRF-TOKEN"] != fixture["csrf_cookie"], label
|
|
assert "2fauth_session" not in browser.cookies, label
|
|
assert fixture["recaller_name"] not in browser.cookies, label
|
|
if driver == "file":
|
|
assert not source.exists(), f"{label}: source PHP file was not destroyed"
|
|
else:
|
|
assert db.execute("SELECT count(*) FROM sessions WHERE id=?", (sid,)).fetchone()[0] == 0, label
|
|
status, user, _ = browser.request("/api/v1/user")
|
|
assert status == 200 and user["id"] == 7, (label, status, user)
|
|
assert db.execute("SELECT user_id FROM rust_sessions").fetchone()[0] == 7, label
|
|
|
|
# Minting a fresh Passport cookie must keep the native browser usable.
|
|
csrf = browser.cookies["XSRF-TOKEN"]
|
|
status, body, _ = browser.request("/oauth/token/refresh", "POST", csrf)
|
|
assert status == 200 and body == "Refreshed.", (label, status, body)
|
|
assert "laravel_token" in browser.cookies, label
|
|
for header in (None, csrf):
|
|
status, user, _ = browser.request("/api/v1/user", csrf=header)
|
|
assert status == 200 and user["id"] == 7, (label, status, user)
|
|
native_before_logout = browser.cookies[NATIVE_COOKIE]
|
|
assert browser.request("/user/logout", "POST", csrf)[0] == 200, label
|
|
assert db.execute("SELECT remember_token FROM users WHERE id=7").fetchone()[0] != "remember-fixture", label
|
|
assert browser.request("/api/v1/user")[0] == 401, label
|
|
replay = Browser(base, original)
|
|
# Exercise the actual import path: the API alone never reads a
|
|
# Laravel session, so an API-only replay would miss a valid recaller.
|
|
assert replay.request("/")[0] == 200, label
|
|
assert replay.request("/api/v1/user", csrf=replay.cookies.get("XSRF-TOKEN"))[0] == 401, f"{label}: PHP session/recaller replay accepted"
|
|
assert Browser(base, {NATIVE_COOKIE: native_before_logout}).request("/api/v1/user")[0] == 401, f"{label}: native session replay accepted"
|
|
return label
|
|
finally:
|
|
db.close()
|
|
process.terminate()
|
|
try:
|
|
process.wait(timeout=5)
|
|
except subprocess.TimeoutExpired:
|
|
process.kill()
|
|
process.wait()
|
|
|
|
|
|
def main():
|
|
binary = Path(sys.argv[1] if len(sys.argv) > 1 else ROOT / "target/debug/twofauth-rs").resolve()
|
|
fixtures = json.loads((ROOT / "tests/fixtures/laravel_sessions.json").read_text())
|
|
passed = [run_case(binary, fixture, driver) for fixture in fixtures for driver in ("file", "database")]
|
|
print(json.dumps({"php_session_browser_migration": "passed", "cases": passed}))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|