Something went wrong. Try again.
Paper Mario mod hosting site - AT Protocol appview starhaven.dev
appview
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448"""Signing up for a Star Haven account, for people without an internet handle."""
import base64import jsonimport reimport threadingimport timeimport uuidfrom concurrent.futures import ThreadPoolExecutorfrom collections.abc import Iteratorfrom pathlib import Pathfrom http.server import BaseHTTPRequestHandler, HTTPServer, ThreadingHTTPServerfrom urllib.parse import parse_qs, quote
import pytestimport requestsfrom conftest import Appview, DevPdsfrom playwright.sync_api import Page, expect
PASSWORD = "Hunter2hunter2"
# Stands in for Turnstile's widget, which would put a solved challenge's# response in the form.TURNSTILE_WIDGET = """for (const widget of document.querySelectorAll(".cf-turnstile")) { const input = document.createElement("input"); input.type = "hidden"; input.name = "cf-turnstile-response"; input.value = "solved"; widget.append(input);}"""
@pytest.fixturedef turnstile_url() -> Iterator[str]: """A stand-in for Turnstile's siteverify, failing only the response "fail"."""
class Handler(BaseHTTPRequestHandler): def do_POST(self) -> None: length = int(self.headers["Content-Length"]) form = parse_qs(self.rfile.read(length).decode()) body = json.dumps({"success": form.get("response") != ["fail"]}).encode() self.send_response(200) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body)
def log_message(self, *args) -> None: pass
server = HTTPServer(("127.0.0.1", 0), Handler) threading.Thread(target=server.serve_forever, daemon=True).start() yield f"http://127.0.0.1:{server.server_port}" server.shutdown()
class PdsProxy: """Passes requests on to the dev PDS, noting each one's path and `X-Forwarded-For`, as the internal address the appview calls it at.
It also lets an admin account's session mint invite codes, as Tranquil PDS does, by trading the session's token for the dev PDS's admin password. Tokens it hasn't seen issued, or that `expire_sessions` has expired, are refused. """
def __init__(self, pds_url: str) -> None: self.forwarded_for: list[tuple[str, str | None]] = [] self.issued: set[str] = set() self.expired: set[str] = set() proxy = self
class Handler(BaseHTTPRequestHandler): def relay(self) -> None: proxy.forwarded_for.append( (self.path.split("?")[0], self.headers.get("X-Forwarded-For")) ) path = self.path.split("?")[0] headers = { name: self.headers[name] for name in ("Content-Type", "Authorization") if name in self.headers } bearer = headers.get("Authorization", "").removeprefix("Bearer ") if path == "/xrpc/com.atproto.server.createInviteCode" and bearer: if bearer in proxy.expired: return self.reply(400, {"error": "ExpiredToken"}) if bearer not in proxy.issued: return self.reply(401, {"error": "AuthenticationRequired"}) headers["Authorization"] = "Basic " + base64.b64encode( b"admin:admin-pass" ).decode()
length = int(self.headers.get("Content-Length") or 0) response = requests.request( self.command, f"{pds_url}{self.path}", data=self.rfile.read(length), headers=headers, timeout=10, ) if response.ok and path in ( "/xrpc/com.atproto.server.createSession", "/xrpc/com.atproto.server.refreshSession", ): proxy.issued.add(response.json()["accessJwt"]) self.send_response(response.status_code) self.send_header("Content-Type", response.headers.get("Content-Type", "")) self.send_header("Content-Length", str(len(response.content))) self.end_headers() self.wfile.write(response.content)
def reply(self, status: int, body: dict) -> None: content = json.dumps(body).encode() self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(content))) self.end_headers() self.wfile.write(content)
do_GET = do_POST = relay
def log_message(self, *args) -> None: pass
self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) self.url = f"http://127.0.0.1:{self.server.server_port}"
def expire_sessions(self) -> None: """Makes every session issued so far expire.""" self.expired |= self.issued
def sent_for(self, path: str) -> list[str | None]: """The `X-Forwarded-For` of every request to `path`.""" return [header for sent, header in self.forwarded_for if sent == path]
@pytest.fixturedef pds_proxy(dev_pds: DevPds) -> Iterator[PdsProxy]: proxy = PdsProxy(dev_pds.pds_url) threading.Thread(target=proxy.server.serve_forever, daemon=True).start() yield proxy proxy.server.shutdown()
@pytest.fixturedef admin_credentials(dev_pds: DevPds, tmp_path: Path) -> Path: """A credentials file for an admin account, as pds.starhaven.dev has.""" handle = f"admin-{uuid.uuid4().hex[:8]}.test" requests.post( f"{dev_pds.pds_url}/xrpc/com.atproto.server.createAccount", json={"handle": handle, "email": f"{handle}@dev.test", "password": PASSWORD}, timeout=10, ).raise_for_status() path = tmp_path / "pds-admin.json" path.write_text(json.dumps({"identifier": handle, "password": PASSWORD})) return path
@pytest.fixturedef server_args( turnstile_url: str, pds_proxy: PdsProxy, admin_credentials: Path) -> list[str]: return [ "--turnstile-verify-url", turnstile_url, "--signup-pds-internal-url", pds_proxy.url, "--pds-admin-credentials-file", str(admin_credentials), ]
def new_name() -> str: """A handle's first part, unused on the dev PDS that every test shares.""" return f"new-{uuid.uuid4().hex[:8]}"
def ask_for_email( appview: Appview, email: str, turnstile: str = "solved", headers: dict[str, str] | None = None, old_enough: bool = True,) -> str: """Submits the first step of sign-up, returning the page it answers with.""" response = requests.post( f"{appview.url}/signup", data={ "email": email, "return_to": "/", "cf-turnstile-response": turnstile, **({"age": "yes"} if old_enough else {}), }, headers=headers, timeout=10, ) response.raise_for_status() return response.text
def finish_links(appview: Appview, email: str) -> list[str]: """The links in every sign-up email the server has printed for `email`.""" return re.findall( rf"email to {re.escape(email)}\n" r"Subject: Finish creating your Star Haven account\n" r".*?(http\S+/signup/finish\?token=\S+)", appview.log.read_text(), re.DOTALL, )
def test_signing_up_creates_an_account_and_signs_in( page: Page, appview: Appview, dev_pds: DevPds): name = new_name() email = f"{name}@dev.test" page.route( "https://challenges.cloudflare.com/**", lambda route: route.fulfill( content_type="text/javascript", body=TURNSTILE_WIDGET ), )
# The appview stands in for an app listening on this computer. app_url = f"{appview.url}/create" page.goto(f"{appview.url}/signup?return_to={quote(app_url, safe='')}") page.get_by_label("Email address").fill(email) page.get_by_label("I'm 13 or older").check() page.get_by_role("button", name="Continue").click() expect(page.get_by_role("heading", name="Check your email")).to_be_visible()
[link] = finish_links(appview, email) page.goto(link) page.get_by_label("Handle").fill(name) page.get_by_label("Password").fill(PASSWORD) page.get_by_role("button", name="Create account").click()
# PDS sign in page.get_by_role("textbox", name="Password").fill(PASSWORD) page.get_by_role("button", name="Sign in").click() page.get_by_role("button", name="Authorize").click()
expect(page).to_have_url(f"{app_url}?handle={name}.test") response = requests.get( f"{dev_pds.pds_url}/xrpc/com.atproto.identity.resolveHandle", params={"handle": f"{name}.test"}, timeout=5, ) response.raise_for_status()
def test_a_sign_up_link_works_once(appview: Appview): name = new_name() email = f"{name}@dev.test" ask_for_email(appview, email) [link] = finish_links(appview, email) token = link.split("token=")[1]
response = requests.post( f"{appview.url}/signup/finish", data={"token": token, "handle": name, "password": PASSWORD}, allow_redirects=False, timeout=10, ) assert response.is_redirect
assert "This link has expired" in requests.get(link, timeout=10).text
def test_a_sign_up_link_makes_one_account_however_often_it_is_sent_at_once( appview: Appview, dev_pds: DevPds): email = f"{new_name()}@dev.test" ask_for_email(appview, email) [link] = finish_links(appview, email) token = link.split("token=")[1] names = [new_name() for _ in range(5)]
def finish(name: str) -> requests.Response: return requests.post( f"{appview.url}/signup/finish", data={"token": token, "handle": name, "password": PASSWORD}, allow_redirects=False, timeout=30, )
with ThreadPoolExecutor(len(names)) as pool: responses = list(pool.map(finish, names))
assert sum(response.is_redirect for response in responses) == 1 created = [ name for name in names if requests.get( f"{dev_pds.pds_url}/xrpc/com.atproto.identity.resolveHandle", params={"handle": f"{name}.test"}, timeout=5, ).ok ] assert len(created) == 1
def test_a_handle_already_taken_asks_for_another(appview: Appview): name = new_name() for email in [f"{name}@dev.test", f"{name}-again@dev.test"]: ask_for_email(appview, email) [first] = finish_links(appview, f"{name}@dev.test") [second] = finish_links(appview, f"{name}-again@dev.test")
for link in [first, second]: response = requests.post( f"{appview.url}/signup/finish", data={ "token": link.split("token=")[1], "handle": name, "password": PASSWORD, }, allow_redirects=False, timeout=10, )
assert f"{name}.test is taken" in response.text
def test_a_failed_security_check_sends_no_email(appview: Appview): email = f"{new_name()}@dev.test"
page = ask_for_email(appview, email, turnstile="fail")
assert "Complete the security check" in page assert finish_links(appview, email) == []
def test_one_address_gets_only_three_emails_a_day(appview: Appview): email = f"{new_name()}@dev.test"
for _ in range(3): assert "Check your email" in ask_for_email(appview, email) page = ask_for_email(appview, email)
assert "sent 3 sign-up emails today" in page assert len(finish_links(appview, email)) == 3
def test_an_address_with_an_account_gets_a_reminder_instead( appview: Appview, dev_pds: DevPds): name = new_name() email = f"{name}@dev.test" requests.post( f"{dev_pds.pds_url}/xrpc/com.atproto.server.createAccount", json={"handle": f"{name}.test", "email": email, "password": PASSWORD}, timeout=10, ).raise_for_status() newcomer = f"{new_name()}@dev.test"
page = ask_for_email(appview, email)
# The same page, but for the address on it, as itself and in the link to # send another. other = ask_for_email(appview, newcomer) other = other.replace(quote(newcomer, safe=""), quote(email, safe="")) assert page == other.replace(newcomer, email) assert finish_links(appview, email) == [] assert ( f"email to {email}\nSubject: You already have a Star Haven account" in appview.log.read_text() )
def test_the_pds_hears_which_visitor_each_sign_up_is_for( appview: Appview, pds_proxy: PdsProxy): name = new_name() email = f"{name}@dev.test" # As a reverse proxy in front of the appview would say. visitor = {"X-Forwarded-For": "203.0.113.7"}
ask_for_email(appview, email, headers=visitor) [link] = finish_links(appview, email) response = requests.post( f"{appview.url}/signup/finish", data={"token": link.split("token=")[1], "handle": name, "password": PASSWORD}, headers=visitor, allow_redirects=False, timeout=10, )
assert response.is_redirect assert pds_proxy.sent_for("/xrpc/_account.checkEmailInUse") == ["203.0.113.7"] assert pds_proxy.sent_for("/xrpc/com.atproto.server.createAccount") == [ "203.0.113.7" ]
def sign_up(appview: Appview) -> None: """Signs up a new account, start to finish.""" name = new_name() email = f"{name}@dev.test" ask_for_email(appview, email) [link] = finish_links(appview, email) response = requests.post( f"{appview.url}/signup/finish", data={"token": link.split("token=")[1], "handle": name, "password": PASSWORD}, allow_redirects=False, timeout=10, ) assert response.is_redirect
def test_the_admin_account_signs_in_once_and_refreshes_its_session( appview: Appview, pds_proxy: PdsProxy): sign_up(appview) sign_up(appview) assert len(pds_proxy.sent_for("/xrpc/com.atproto.server.createSession")) == 1 assert pds_proxy.sent_for("/xrpc/com.atproto.server.refreshSession") == []
# The dev PDS's tokens differ only by when they were issued, to the # second, so a refresh in the same second would mint one that's expired. time.sleep(1.1) pds_proxy.expire_sessions() sign_up(appview)
assert len(pds_proxy.sent_for("/xrpc/com.atproto.server.createSession")) == 1 assert len(pds_proxy.sent_for("/xrpc/com.atproto.server.refreshSession")) == 1
def test_sign_up_needs_someone_13_or_older(appview: Appview): email = f"{new_name()}@dev.test"
page = ask_for_email(appview, email, old_enough=False)
assert "Confirm you're 13 or older" in page assert finish_links(appview, email) == []
def test_the_policies_open_beside_the_sign_up_form(page: Page, appview: Appview): page.goto(f"{appview.url}/signup")
for name in ["terms of service", "privacy policy"]: link = page.locator("form").get_by_role("link", name=name) expect(link).to_have_attribute("target", "_blank") expect(link).to_have_attribute("rel", "noopener")