Merge branch 'develop' into bma/settingsIterationStep2Bis

tools/diawi/install_pr_build.py

@@ -0,0 +1,237 @@
+#!/usr/bin/env python3
+"""
+Install the Diawi build attached to a GitHub PR onto a connected Android device.
+
+Usage:
+    ./install_pr_build.py https://github.com/element-hq/element-x-android/pull/7750
+    ./install_pr_build.py <pr_url> --serial emulator-5554 --replace
+
+Steps:
+  1. Reads the PR description and comments through the GitHub API and finds the
+     most recent link of the form https://i.diawi.com/<build_id>.
+  2. Scrapes the Diawi page and finds the link labeled "Download application".
+  3. Downloads the APK.
+  4. Runs `adb install <apk_path>`.
+
+Set GITHUB_TOKEN in the environment to avoid GitHub's anonymous rate limit
+(60 requests/hour). Only the Python standard library is used.
+"""
+
+import argparse
+import json
+import os
+import re
+import shutil
+import subprocess
+import sys
+import tempfile
+import urllib.error
+import urllib.request
+from html.parser import HTMLParser
+from urllib.parse import urljoin
+
+PR_URL_RE = re.compile(r"https?://github\.com/([^/]+)/([^/]+)/pull/(\d+)")
+DIAWI_RE = re.compile(r"https://i\.diawi\.com/([A-Za-z0-9]+)")
+DOWNLOAD_LABEL = "download application"
+USER_AGENT = (
+    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
+    "(KHTML, like Gecko) Chrome/128.0 Safari/537.36"
+)
+
+
+def die(msg: str) -> None:
+    print(f"error: {msg}", file=sys.stderr)
+    sys.exit(1)
+
+
+def http_get(url: str, headers: dict | None = None) -> urllib.request.addinfourl:
+    req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT, **(headers or {})})
+    try:
+        return urllib.request.urlopen(req, timeout=60)
+    except urllib.error.HTTPError as e:
+        die(f"HTTP {e.code} while fetching {url}")
+    except urllib.error.URLError as e:
+        die(f"could not reach {url}: {e.reason}")
+
+
+# --------------------------------------------------------------------------- #
+# 1. Find the Diawi link in the PR
+# --------------------------------------------------------------------------- #
+def github_json(url: str):
+    headers = {"Accept": "application/vnd.github+json"}
+    token = os.environ.get("GITHUB_TOKEN")
+    if token:
+        headers["Authorization"] = f"Bearer {token}"
+    with http_get(url, headers) as resp:
+        return json.load(resp), resp.headers.get("Link", "")
+
+
+def find_diawi_link(pr_url: str) -> str:
+    m = PR_URL_RE.match(pr_url)
+    if not m:
+        die(f"not a GitHub PR URL: {pr_url}")
+    owner, repo, number = m.groups()
+    api = f"https://api.github.com/repos/{owner}/{repo}"
+
+    texts = []
+    pr, _ = github_json(f"{api}/pulls/{number}")
+    texts.append(pr.get("body") or "")
+
+    # Comments are returned oldest first; follow pagination to get them all.
+    url = f"{api}/issues/{number}/comments?per_page=100"
+    while url:
+        comments, link_header = github_json(url)
+        texts.extend(c.get("body") or "" for c in comments)
+        nxt = re.search(r'<([^>]+)>;\s*rel="next"', link_header)
+        url = nxt.group(1) if nxt else None
+
+    links = [f"https://i.diawi.com/{bid}" for t in texts for bid in DIAWI_RE.findall(t)]
+    if not links:
+        die("no https://i.diawi.com/<build_id> link found in the PR")
+    return links[-1]  # most recent one
+
+
+# --------------------------------------------------------------------------- #
+# 2. Scrape the Diawi page for the "Download application" link
+# --------------------------------------------------------------------------- #
+class LinkCollector(HTMLParser):
+    """Collects (href, visible text) for every <a> tag."""
+
+    def __init__(self):
+        super().__init__()
+        self.links: list[tuple[str, str]] = []
+        self._href: str | None = None
+        self._text: list[str] = []
+
+    def handle_starttag(self, tag, attrs):
+        if tag == "a":
+            self._href = dict(attrs).get("href")
+            self._text = []
+
+    def handle_data(self, data):
+        if self._href is not None:
+            self._text.append(data)
+
+    def handle_endtag(self, tag):
+        if tag == "a" and self._href is not None:
+            text = " ".join("".join(self._text).split())
+            self.links.append((self._href, text))
+            self._href = None
+
+
+def find_apk_url(diawi_url: str) -> str:
+    with http_get(diawi_url) as resp:
+        html = resp.read().decode(resp.headers.get_content_charset() or "utf-8", "replace")
+
+    parser = LinkCollector()
+    parser.feed(html)
+
+    for href, text in parser.links:
+        if href and DOWNLOAD_LABEL in text.lower():
+            return urljoin(diawi_url, href)
+
+    # Fallback: anything that looks like a direct APK link.
+    for href, _ in parser.links:
+        if href and ".apk" in href.lower():
+            return urljoin(diawi_url, href)
+
+    die(f"no 'Download application' link found on {diawi_url} "
+        "(the build may have expired)")
+
+
+# --------------------------------------------------------------------------- #
+# 3. Download the APK
+# --------------------------------------------------------------------------- #
+def download(url: str, dest: str) -> None:
+    with http_get(url) as resp, open(dest, "wb") as out:
+        total = int(resp.headers.get("Content-Length") or 0)
+        done = 0
+        while chunk := resp.read(1 << 16):
+            out.write(chunk)
+            done += len(chunk)
+            if total:
+                print(f"\r  {done * 100 // total:3d}%  {done / 1e6:.1f} MB", end="", flush=True)
+    print()
+
+    with open(dest, "rb") as f:
+        if f.read(2) != b"PK":  # APKs are zip archives
+            die("downloaded file is not an APK (Diawi may have returned an HTML page)")
+
+
+# --------------------------------------------------------------------------- #
+# 4. Install with adb
+# --------------------------------------------------------------------------- #
+def adb_install(apk: str, serial: str | None, replace: bool) -> int:
+    if not shutil.which("adb"):
+        die("adb not found in PATH")
+    adb = ["adb"] + (["-s", serial] if serial else [])
+
+    devices = subprocess.run(["adb", "devices"], capture_output=True, text=True).stdout
+    attached = [l for l in devices.splitlines()[1:] if l.strip().endswith("device")]
+    if not attached:
+        die("no Android device connected (check `adb devices`)")
+
+    cmd = adb + ["install"] + (["-r"] if replace else []) + [apk]
+    print("Running:", " ".join(cmd))
+    return subprocess.run(cmd).returncode
+
+def adb_launch(serial: str | None) -> int:
+    if not shutil.which("adb"):
+        die("adb not found in PATH")
+    adb = ["adb"] + (["-s", serial] if serial else [])
+
+    devices = subprocess.run(["adb", "devices"], capture_output=True, text=True).stdout
+    attached = [l for l in devices.splitlines()[1:] if l.strip().endswith("device")]
+    if not attached:
+        die("no Android device connected (check `adb devices`)")
+
+    cmd = adb + ["shell", "am", "start", "-n", "io.element.android.x.debug/io.element.android.x.MainActivity"]
+
+    print("Running:", " ".join(cmd))
+    return subprocess.run(cmd).returncode
+
+def main() -> None:
+    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    ap.add_argument("pr_url", help="GitHub PR URL or PR number (e.g. 7750)")
+    ap.add_argument("-s", "--serial", help="device serial, if several devices are connected")
+    ap.add_argument("-r", "--replace", action="store_true", help="pass -r to adb install (reinstall, keep data)")
+    ap.add_argument("-o", "--output", help="where to save the APK (default: temp dir, deleted afterwards)")
+    args = ap.parse_args()
+
+    if args.pr_url.isdigit():
+        pr_url = f"https://github.com/element-hq/element-x-android/pull/{args.pr_url}"
+    else:
+        pr_url = args.pr_url
+    print(f"Looking for a Diawi link in {pr_url} ...")
+    diawi_url = find_diawi_link(pr_url)
+    print(f"Found: {diawi_url}")
+
+    apk_url = find_apk_url(diawi_url)
+    print(f"APK link: {apk_url}")
+
+    build_id = diawi_url.rsplit("/", 1)[-1]
+    if args.output:
+        apk_path = args.output
+        tmpdir = None
+    else:
+        tmpdir = tempfile.mkdtemp(prefix="diawi-")
+        apk_path = os.path.join(tmpdir, f"{build_id}.apk")
+
+    print(f"Downloading to {apk_path} ...")
+    try:
+        download(apk_url, apk_path)
+        if adb_install(apk_path, args.serial, args.replace) == 0:
+            print("Installed successfully.")
+        else:
+            die("adb install failed.")
+        if adb_launch(args.serial) == 0:
+            print("Launched successfully.")
+        else:
+            die("adb launch failed.")
+    finally:
+        if tmpdir:
+            shutil.rmtree(tmpdir, ignore_errors=True)
+
+
+if __name__ == "__main__":
+    main()

tools/mas/delete_all_sessions.py

@@ -0,0 +1,296 @@
+#!/usr/bin/env python3
+
+# Copyright (c) 2026 Element Creations Ltd.
+#
+# SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial.
+
+"""End all the sessions of a Matrix account, using MAS (Matrix Authentication Service).
+
+The script logs in to MAS using the web login form (username and password), and then uses
+the MAS GraphQL API with the obtained browser session cookie to list and end all the active
+compatibility sessions, OAuth 2.0 sessions and browser sessions of the account.
+
+Requirements:
+- MAS password login must be enabled, with no captcha on the login form.
+- Python package `requests`.
+"""
+
+from __future__ import annotations
+
+import argparse
+import getpass
+import os
+import re
+import sys
+import time
+from dataclasses import dataclass
+
+import requests
+
+PAGE_SIZE = 100
+# Timeout in seconds for each HTTP request, so that the script never hangs forever.
+TIMEOUT = 30
+
+SESSION_TYPES = {
+    # type: (User field, node fields, mutation, input field)
+    "compat": ("compatSessions", "id deviceId lastActiveAt", "endCompatSession", "compatSessionId"),
+    "oauth2": ("oauth2Sessions", "id lastActiveAt client { clientName clientId }", "endOauth2Session", "oauth2SessionId"),
+    "browser": ("browserSessions", "id lastActiveAt userAgent { raw }", "endBrowserSession", "browserSessionId"),
+}
+
+
+@dataclass
+class MasSession:
+    type: str
+    id: str
+    description: str
+    last_active_at: str | None
+
+
+# Set by the --verbose option.
+verbose_logs = False
+
+
+def log(message: str, verbose: bool = False) -> None:
+    """Log a message. Verbose messages are only displayed with the --verbose option."""
+    if verbose and not verbose_logs:
+        return
+    print(f"[{time.strftime('%H:%M:%S')}] {message}", flush=True)
+
+
+class MasClient:
+    def __init__(self, mas_url: str):
+        self.mas_url = mas_url.rstrip("/")
+        self.http = requests.Session()
+
+    def login(self, username: str, password: str) -> None:
+        log(f"Fetching the login form from {self.mas_url}/login", verbose=True)
+        response = self.http.get(f"{self.mas_url}/login", timeout=TIMEOUT)
+        log(f"Got HTTP {response.status_code}", verbose=True)
+        response.raise_for_status()
+        match = re.search(r'name="csrf"\s+value="([^"]+)"', response.text) or re.search(r'value="([^"]+)"\s+name="csrf"', response.text)
+        if not match:
+            raise RuntimeError("Could not find the CSRF token in the MAS login form. Is password login enabled?")
+        log(f"Submitting the login form for {username}", verbose=True)
+        response = self.http.post(
+            f"{self.mas_url}/login",
+            data={"csrf": match.group(1), "username": username, "password": password},
+            allow_redirects=False,
+            timeout=TIMEOUT,
+        )
+        log(f"Got HTTP {response.status_code}", verbose=True)
+        # On success MAS redirects, on failure it renders the login form again.
+        if not response.is_redirect:
+            raise RuntimeError(f"Login failed (HTTP {response.status_code}). Check the username and password.")
+        log(f"Logged in, redirected to {response.headers.get('location')}")
+
+    def graphql(self, query: str, variables: dict | None = None) -> dict:
+        response = self.http.post(f"{self.mas_url}/graphql", json={"query": query, "variables": variables or {}}, timeout=TIMEOUT)
+        log(f"GraphQL request done: HTTP {response.status_code}", verbose=True)
+        response.raise_for_status()
+        body = response.json()
+        if body.get("errors"):
+            raise RuntimeError(f"GraphQL error: {body['errors']}")
+        return body["data"]
+
+    def current_browser_session_id(self) -> str | None:
+        log("Fetching the current browser session", verbose=True)
+        data = self.graphql("query { viewerSession { __typename ... on BrowserSession { id } } }")
+        return data["viewerSession"].get("id")
+
+    def list_sessions(self, session_type: str) -> tuple[list[MasSession], bool]:
+        """Return the active sessions of the given type, and whether the list is complete."""
+        field, node_fields, _, _ = SESSION_TYPES[session_type]
+        query = f"""
+            query ($first: Int!, $after: String) {{
+              viewer {{
+                __typename
+                ... on User {{
+                  {field}(state: ACTIVE, first: $first, after: $after) {{
+                    edges {{ node {{ {node_fields} }} }}
+                    pageInfo {{ hasNextPage endCursor }}
+                  }}
+                }}
+              }}
+            }}
+        """
+        sessions: dict[str, MasSession] = {}
+        after = None
+        while True:
+            log(f"Fetching a page of active {session_type} sessions after cursor {after}", verbose=True)
+            viewer = self.graphql(query, {"first": PAGE_SIZE, "after": after})["viewer"]
+            if viewer["__typename"] != "User":
+                raise RuntimeError("Not logged in to MAS.")
+            connection = viewer[field]
+            new_count = 0
+            for edge in connection["edges"]:
+                session = to_mas_session(session_type, edge["node"])
+                if session.id not in sessions:
+                    sessions[session.id] = session
+                    new_count += 1
+            page_info = connection["pageInfo"]
+            log(
+                f"Got {len(connection['edges'])} {session_type} session(s), {new_count} new, {len(sessions)} found so far, "
+                f"hasNextPage: {page_info['hasNextPage']}, endCursor: {page_info['endCursor']}",
+                verbose=True,
+            )
+            if not page_info["hasNextPage"]:
+                log(f"Found {len(sessions)} active {session_type} session(s)", verbose=True)
+                return list(sessions.values()), True
+            # MAS can return the same results again for the next page (seen with browser sessions), which would loop forever.
+            if new_count == 0 or page_info["endCursor"] in (None, after):
+                log(f"Pagination of {session_type} sessions is not progressing, stopping at {len(sessions)} session(s)", verbose=True)
+                return list(sessions.values()), False
+            after = page_info["endCursor"]
+
+    def list_all_sessions(self) -> tuple[list[MasSession], bool]:
+        """Return the active sessions of all types, and whether the list is complete."""
+        all_sessions = []
+        all_complete = True
+        for session_type in SESSION_TYPES:
+            sessions, complete = self.list_sessions(session_type)
+            all_sessions += sessions
+            all_complete = all_complete and complete
+        return all_sessions, all_complete
+
+    def end_session(self, session: MasSession) -> str:
+        _, _, mutation, input_field = SESSION_TYPES[session.type]
+        query = f"""
+            mutation ($id: ID!) {{
+              {mutation}(input: {{ {input_field}: $id }}) {{ status }}
+            }}
+        """
+        log(f"Ending {session.type} session {session.id} ({session.description})", verbose=True)
+        status = self.graphql(query, {"id": session.id})[mutation]["status"]
+        log(f"Status: {status}", verbose=True)
+        return status
+
+
+def to_mas_session(session_type: str, node: dict) -> MasSession:
+    if session_type == "compat":
+        description = f"device {node.get('deviceId')}"
+    elif session_type == "oauth2":
+        client = node.get("client") or {}
+        description = client.get("clientName") or client.get("clientId") or "unknown client"
+    else:
+        description = (node.get("userAgent") or {}).get("raw") or "unknown user agent"
+    return MasSession(session_type, node["id"], description, node.get("lastActiveAt"))
+
+
+def discover_mas_url(homeserver: str) -> str:
+    server = homeserver if homeserver.startswith("http") else f"https://{homeserver}"
+    server = server.rstrip("/")
+    base_url = server
+    log(f"Fetching {server}/.well-known/matrix/client", verbose=True)
+    try:
+        response = requests.get(f"{server}/.well-known/matrix/client", timeout=10)
+        if response.ok:
+            base_url = response.json().get("m.homeserver", {}).get("base_url", server).rstrip("/")
+        log(f"Got HTTP {response.status_code}, homeserver base URL: {base_url}", verbose=True)
+    except (requests.RequestException, ValueError) as e:
+        log(f"Could not read the .well-known file ({e}), using {base_url}", verbose=True)
+    for path in ("/_matrix/client/v1/auth_metadata", "/_matrix/client/unstable/org.matrix.msc2965/auth_metadata"):
+        log(f"Fetching {base_url}{path}", verbose=True)
+        response = requests.get(f"{base_url}{path}", timeout=10)
+        log(f"Got HTTP {response.status_code}", verbose=True)
+        if response.ok:
+            return response.json()["issuer"]
+    raise RuntimeError(f"Could not discover the MAS URL from {base_url}. Is the homeserver using MAS? Use --mas-url otherwise.")
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="End all the sessions of a Matrix account, using MAS.")
+    parser.add_argument("--homeserver", help="Homeserver name or URL, e.g. matrix.example.org. Used to discover the MAS URL.")
+    parser.add_argument("--mas-url", help="MAS URL, e.g. https://account.example.org. Overrides the discovery.")
+    parser.add_argument("--username", required=True, help="Username (localpart) of the account.")
+    parser.add_argument("--password", help="Password of the account. Defaults to the MAS_PASSWORD env variable, or prompted.")
+    parser.add_argument("--dry-run", action="store_true", help="Only list the sessions, do not end them.")
+    parser.add_argument("--keep-browser-session", action="store_true", help="Do not end the browser session created by this script.")
+    parser.add_argument("--verbose", action="store_true", help="Log every request and every session, and list all the sessions.")
+    args = parser.parse_args()
+
+    global verbose_logs
+    verbose_logs = args.verbose
+
+    if not args.homeserver and not args.mas_url:
+        parser.error("one of --homeserver or --mas-url is required")
+
+    password = args.password or os.environ.get("MAS_PASSWORD") or getpass.getpass(f"Password for {args.username}: ")
+
+    try:
+        mas_url = args.mas_url or discover_mas_url(args.homeserver)
+        log(f"Using MAS at {mas_url}")
+        client = MasClient(mas_url)
+        client.login(args.username, password)
+        own_session_id = client.current_browser_session_id()
+        if args.dry_run:
+            sessions, complete = client.list_all_sessions()
+    except (requests.RequestException, RuntimeError) as e:
+        print(f"Error: {e}", file=sys.stderr)
+        return 1
+
+    if args.dry_run:
+        log(f"Found {len(sessions)} active session(s)")
+        for session in sessions:
+            own = " (this script)" if session.id == own_session_id else ""
+            print(f"  {session.type:<8} {session.id}  last active: {session.last_active_at or '-':<30} {session.description}{own}")
+        if not complete:
+            log("Warning: MAS pagination is not progressing, so the list is incomplete. More sessions will be found once these are ended.")
+        return 0
+
+    # MAS pagination does not always progress, so fetch the sessions, end them, and repeat until none are left.
+    # The session used by this script is ended at the end, so that it can end all the others.
+    ended = 0
+    failed = 0
+    attempted: set[str] = set()
+    batch_number = 0
+    while True:
+        batch_number += 1
+        try:
+            sessions, complete = client.list_all_sessions()
+        except (requests.RequestException, RuntimeError) as e:
+            print(f"Error: {e}", file=sys.stderr)
+            failed += 1
+            break
+        batch = [session for session in sessions if session.id != own_session_id and session.id not in attempted]
+        if not batch:
+            if not complete:
+                log("Warning: stopping, the remaining sessions cannot be fetched because the previous ones could not be ended")
+            break
+        log(f"Batch {batch_number}: ending {len(batch)} session(s)" + ("" if complete else ", more will be fetched afterwards"))
+        for session in batch:
+            attempted.add(session.id)
+            if args.verbose:
+                print(f"  {session.type:<8} {session.id}  last active: {session.last_active_at or '-':<30} {session.description}")
+            if end_session(client, session):
+                ended += 1
+            else:
+                failed += 1
+            if (ended + failed) % PAGE_SIZE == 0:
+                log(f"Progress: {ended + failed} session(s) processed, {failed} failure(s)")
+        if complete:
+            break
+
+    if own_session_id:
+        if args.keep_browser_session:
+            log(f"Keeping browser session {own_session_id} used by this script")
+        elif end_session(client, MasSession("browser", own_session_id, "used by this script", None)):
+            ended += 1
+        else:
+            failed += 1
+
+    print(f"Ended {ended} session(s), {failed} failure(s).")
+    return 0 if failed == 0 else 1
+
+
+def end_session(client: MasClient, session: MasSession) -> bool:
+    try:
+        status = client.end_session(session)
+    except (requests.RequestException, RuntimeError) as e:
+        status = str(e)
+    if status != "ENDED":
+        print(f"Failed to end {session.type} session {session.id}: {status}", file=sys.stderr)
+    return status == "ENDED"
+
+if __name__ == "__main__":
+    sys.exit(main())