Matrix Administration

{
  "skill_name": "matrix-administration",
  "evals": [
    {
      "id": 1,
      "eval_name": "snapshot-rooms",
      "prompt": "Snapshot every room on the configured Synapse homeserver into rooms.json.",
      "expected_output": "Runs synapse-fetch-rooms.py.",
      "files": [],
      "assertions": [
        "Invokes synapse-fetch-rooms.py via python3",
        "Does not pass --server unless the user specified a server suffix",
        "Does not commit or print the resulting rooms.json"
      ]
    },
    {
      "id": 2,
      "eval_name": "rate-rooms-de",
      "prompt": "Rate the rooms in rooms.json with German output, treating !home:example.com as our home space.",
      "expected_output": "Runs synapse-rate-rooms.py with --language de --space '!home:example.com'.",
      "files": [],
      "assertions": [
        "Invokes synapse-rate-rooms.py",
        "Passes --language de (or sets LANGUAGE=de)",
        "Passes --space '!home:example.com'"
      ]
    },
    {
      "id": 3,
      "eval_name": "render-graph",
      "prompt": "Render an SVG of the room tree.",
      "expected_output": "Runs synapse-graph.py and uses any home_space_ids from the config or asks the user.",
      "files": [],
      "assertions": [
        "Invokes synapse-graph.py",
        "Surfaces that graphviz `dot` is required for the SVG step (or uses --no-svg)"
      ]
    },
    {
      "id": 4,
      "eval_name": "deactivate-user-confirm",
      "prompt": "Deactivate @leaver:example.com.",
      "expected_output": "Confirms before running synapse-deactivate-user.py because the operation is irreversible.",
      "files": [],
      "assertions": [
        "Surfaces that the operation is irreversible",
        "Does NOT silently pass --yes",
        "Asks the user about --erase (GDPR removal of messages)"
      ]
    },
    {
      "id": 5,
      "eval_name": "harden-room",
      "prompt": "Make !room:example.com private, add it to !home:example.com, and enable encryption. Use @admin:example.com as the operator.",
      "expected_output": "Runs synapse-migrate-room.py with all three IDs.",
      "files": [],
      "assertions": [
        "Invokes synapse-migrate-room.py",
        "Passes the room, user and space IDs in the correct order",
        "Surfaces that enabling encryption is irreversible"
      ]
    },
    {
      "id": 6,
      "eval_name": "audit-single-admin-rooms",
      "prompt": "Find rooms where @alice:example.com is the only admin.",
      "expected_output": "Uses rooms.json + synapse-user-admin-rooms.py to filter rooms with no other admins.",
      "files": [],
      "assertions": [
        "Snapshots first if rooms.json is missing/stale",
        "Invokes synapse-user-admin-rooms.py",
        "Highlights rooms with '(no other admins)'"
      ]
    },
    {
      "id": 7,
      "eval_name": "search-encrypted-warning",
      "prompt": "Search !devops:example.com for messages from @bot:example.com containing 'deploy'.",
      "expected_output": "Runs synapse-search.py and warns that E2EE messages are invisible.",
      "files": [],
      "assertions": [
        "Invokes synapse-search.py",
        "Warns that empty results do not mean no messages if the room is encrypted"
      ]
    },
    {
      "id": 8,
      "eval_name": "biggest-rooms-default",
      "prompt": "What are the 5 biggest rooms on the homeserver by DB size?",
      "expected_output": "Runs synapse-biggest-rooms.py with -n 5.",
      "files": [],
      "assertions": [
        "Invokes synapse-biggest-rooms.py",
        "Passes -n 5 (or --limit 5)",
        "Does not require a snapshot — calls the live admin API"
      ]
    },
    {
      "id": 9,
      "eval_name": "member-flow-with-server-filter",
      "prompt": "Show the join/leave timeline of !ops:example.com and highlight users from example.com.",
      "expected_output": "Runs synapse-room-member-flow.py with the room ID and --server :example.com.",
      "files": [],
      "assertions": [
        "Invokes synapse-room-member-flow.py",
        "Passes --server :example.com (or relies on room_filter from config)"
      ]
    },
    {
      "id": 10,
      "eval_name": "add-room-to-space",
      "prompt": "Link !room:example.com into !home:example.com as a child.",
      "expected_output": "Runs synapse-add-to-space.py with both IDs (or relies on default_space_id from config).",
      "files": [],
      "assertions": [
        "Invokes synapse-add-to-space.py",
        "Passes the room ID first, then the space ID",
        "Does not enable encryption or change join rules — that's the migrate script's job"
      ]
    }
  ]
}

"""matrix-administration shared library.

Stdlib only.  At the top of each script:

    import sys, os
    sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
    from _lib import load_config, admin_request, client_request
"""

from _lib.admin_http import admin_request, client_request, quote
from _lib.colors import bold, cyan, gray, green, red, yellow
from _lib.condensing import Room, condense
from _lib.config import load_config
from _lib.pretty_bytes import pretty_bytes
from _lib.rating import RoomRating, format_rating, rate_room, rating_emoji

__all__ = [
    "admin_request",
    "client_request",
    "quote",
    "bold",
    "cyan",
    "gray",
    "green",
    "red",
    "yellow",
    "Room",
    "condense",
    "load_config",
    "pretty_bytes",
    "RoomRating",
    "format_rating",
    "rate_room",
    "rating_emoji",
]

"""HTTP wrappers for the Synapse Admin API and Matrix Client-Server API.

Both admin endpoints (under ``/_synapse/admin``) and standard Matrix client
endpoints (under ``/_matrix/client/v3``) are reachable through the same HTTP
helpers.  Most helpers in this module accept an admin token; a few stubs are
provided for endpoints that require client-server semantics.

Stdlib only.
"""

from __future__ import annotations

import contextlib
import json
import socket
import urllib.error
import urllib.parse
import urllib.request


_ALLOWED_SCHEMES = frozenset({"http", "https"})


def _require_http_scheme(url: str) -> None:
    scheme = urllib.parse.urlparse(url).scheme
    if scheme not in _ALLOWED_SCHEMES:
        raise ValueError(
            f"Refusing to fetch URL with scheme {scheme!r}; only http/https allowed"
        )


@contextlib.contextmanager
def _prefer_ipv4():
    """Temporarily prefer IPv4 in DNS resolution (WSL2 workaround)."""
    original = socket.getaddrinfo

    def patched(*args, **kwargs):
        results = original(*args, **kwargs)
        return sorted(results, key=lambda r: r[0] != socket.AF_INET)

    socket.getaddrinfo = patched
    try:
        yield
    finally:
        socket.getaddrinfo = original


_DEFAULT_TIMEOUT = 60


def _do_request(req: urllib.request.Request) -> dict:
    """Execute a request and return parsed JSON response.

    The caller must have validated the URL scheme via
    ``_require_http_scheme`` before constructing ``req``.  A 60-second
    timeout is applied so a hung server does not freeze the script.
    """
    _require_http_scheme(req.full_url)
    # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected
    with urllib.request.urlopen(req, timeout=_DEFAULT_TIMEOUT) as response:  # noqa: S310 — scheme validated above
        body = response.read().decode()
        if not body:
            return {}
        return json.loads(body)


def _parse_http_error(e: urllib.error.HTTPError) -> dict:
    error_body = e.read().decode()
    try:
        parsed = json.loads(error_body)
        return {
            "error": parsed.get("error", error_body),
            "errcode": parsed.get("errcode"),
            "status": e.code,
        }
    except json.JSONDecodeError:
        return {"error": error_body, "errcode": str(e.code), "status": e.code}


def _request(url: str, method: str, token: str, data: dict | None) -> dict:
    _require_http_scheme(url)
    headers = {"Authorization": f"Bearer {token}"}
    body = None
    if data is not None:
        headers["Content-Type"] = "application/json"
        body = json.dumps(data).encode()

    req = urllib.request.Request(url, data=body, headers=headers, method=method)
    try:
        return _do_request(req)
    except urllib.error.HTTPError as e:
        return _parse_http_error(e)
    except OSError as e:
        if e.errno not in (101, 113):
            return {"error": str(e)}
        try:
            with _prefer_ipv4():
                req2 = urllib.request.Request(
                    url, data=body, headers=headers, method=method
                )
                return _do_request(req2)
        except urllib.error.HTTPError as e2:
            return _parse_http_error(e2)
        except OSError as e2:
            return {"error": str(e2)}


def admin_request(
    config: dict, method: str, endpoint: str, data: dict | None = None
) -> dict:
    """Call a Synapse Admin API endpoint.

    Args:
        config: Dict with ``homeserver`` and ``admin_token`` (or ``access_token``).
        method: HTTP method.
        endpoint: Admin path beginning with ``/v1/...`` or ``/v2/...``
            (the ``/_synapse/admin`` prefix is added automatically).
        data: JSON body, if any.

    Returns:
        Parsed JSON response, or a dict with ``error``/``errcode`` on failure.
    """
    token = config.get("admin_token") or config["access_token"]
    url = f"{config['homeserver']}/_synapse/admin{endpoint}"
    return _request(url, method, token, data)


def client_request(
    config: dict, method: str, endpoint: str, data: dict | None = None
) -> dict:
    """Call a Matrix Client-Server v3 endpoint with the admin token.

    Used for state events and other Matrix-spec calls that the admin token
    can perform (the admin user must be in the room).

    Args:
        config: Dict with ``homeserver`` and ``admin_token`` (or ``access_token``).
        method: HTTP method.
        endpoint: Path beginning with ``/`` (the ``/_matrix/client/v3`` prefix
            is added automatically).
        data: JSON body, if any.
    """
    token = config.get("admin_token") or config["access_token"]
    url = f"{config['homeserver']}/_matrix/client/v3{endpoint}"
    return _request(url, method, token, data)


def quote(value: str) -> str:
    """URL-encode a path segment (room IDs, user IDs, aliases)."""
    return urllib.parse.quote(value, safe="")

"""Minimal ANSI colour helpers.

Disabled automatically when stdout is not a TTY or ``NO_COLOR`` is set.
"""

from __future__ import annotations

import os
import sys


def _enabled() -> bool:
    if os.environ.get("NO_COLOR"):
        return False
    return sys.stdout.isatty()


_CODES = {
    "reset": "\x1b[0m",
    "bold": "\x1b[1m",
    "gray": "\x1b[90m",
    "red": "\x1b[31m",
    "green": "\x1b[32m",
    "yellow": "\x1b[33m",
    "blue": "\x1b[34m",
    "magenta": "\x1b[35m",
    "cyan": "\x1b[36m",
}


def _wrap(text: str, name: str) -> str:
    if not _enabled():
        return text
    return f"{_CODES[name]}{text}{_CODES['reset']}"


def bold(text: str) -> str:
    return _wrap(text, "bold")


def gray(text: str) -> str:
    return _wrap(text, "gray")


def red(text: str) -> str:
    return _wrap(text, "red")


def green(text: str) -> str:
    return _wrap(text, "green")


def yellow(text: str) -> str:
    return _wrap(text, "yellow")


def blue(text: str) -> str:
    return _wrap(text, "blue")


def cyan(text: str) -> str:
    return _wrap(text, "cyan")

"""Reduce the verbose ``rooms.json`` snapshot into a small ``Room`` graph.

Mirrors the original ``lib/condensing.mjs`` from the matrix-tools project
but is fully Python/stdlib.  See ``synapse-fetch-rooms.py`` for the input
format (a list of ``{"room": …, "states": [state events …]}`` entries).

Stdlib only.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Optional


# A join policy is one of:
#   "public"
#   "invite-only"
#   "unknown"
#   ("in-space", [allow_room_id, …])  — room version 9+ restricted joins
JoinPolicy = object  # documented above; kept as object for simplicity


@dataclass
class Room:
    id: str
    name: str
    version: int
    creator: Optional[str] = None
    join_policy: JoinPolicy = "public"
    space_children: list[str] = field(default_factory=list)
    permissions: dict[str, int] = field(default_factory=dict)
    members: dict[str, datetime] = field(default_factory=dict)
    is_encrypted: bool = False
    is_space: bool = False
    is_replaced: bool = False
    predecessor: Optional[str] = None


def condense(data: list[dict]) -> dict[str, Room]:
    """Convert a list of fetched rooms with their state events into a dict
    of ``Room`` objects keyed by room ID.

    Rooms with no human-readable name are skipped (they are usually direct
    messages or internal rooms with no meaningful identity).
    """
    rooms: dict[str, Room] = {}

    for entry in data:
        room = entry.get("room", {})
        states = entry.get("states", [])

        name = room.get("name")
        if not name:
            continue

        is_space = room.get("room_type") == "m.space"
        try:
            version = int(room.get("version", 0))
        except (TypeError, ValueError):
            version = 0

        r = Room(
            id=room["room_id"],
            name=name,
            version=version,
            is_encrypted=room.get("encryption") is not None,
            is_space=is_space,
        )
        rooms[r.id] = r

        for state in states:
            stype = state.get("type")
            content = state.get("content", {}) or {}

            if stype == "m.room.create":
                r.creator = content.get("creator") or state.get("sender")
                pred = content.get("predecessor")
                if isinstance(pred, dict) and pred.get("room_id"):
                    r.predecessor = pred["room_id"]

            elif stype == "m.room.tombstone" and content.get("replacement_room"):
                r.is_replaced = True

            elif stype == "m.space.child" and content:
                r.space_children.append(state.get("state_key", ""))

            elif stype == "m.room.join_rules":
                rule = content.get("join_rule")
                if rule == "public":
                    r.join_policy = "public"
                elif rule == "invite":
                    r.join_policy = "invite-only"
                elif rule == "restricted":
                    allow = content.get("allow") or []
                    room_ids = [
                        a["room_id"]
                        for a in allow
                        if isinstance(a, dict)
                        and a.get("type") == "m.room_membership"
                        and a.get("room_id")
                    ]
                    r.join_policy = ("in-space", room_ids)
                else:
                    r.join_policy = "unknown"

            elif stype == "m.room.power_levels":
                users = content.get("users") or {}
                for uid, level in users.items():
                    try:
                        r.permissions[uid] = int(level)
                    except (TypeError, ValueError):
                        continue

            elif (
                stype == "m.room.member"
                and content.get("membership") == "join"
                and state.get("state_key")
            ):
                ts = state.get("origin_server_ts")
                if isinstance(ts, (int, float)):
                    r.members[state["state_key"]] = datetime.fromtimestamp(
                        ts / 1000, tz=timezone.utc
                    )

    return rooms

"""Configuration loading for Synapse Admin scripts.

Reuses ``~/.config/matrix/config.json`` (the same file the
``matrix-communication`` skill reads).  Admin scripts require an admin-level
token and accept the following optional fields in addition to the standard
ones:

- ``admin_token``: a Matrix access token for a user with Synapse server-admin
  rights.  Falls back to ``access_token`` if absent.
- ``room_filter``: optional server-suffix filter applied by
  ``synapse-fetch-rooms.py`` (e.g. ``":example.com"``).  Empty/missing means
  no filter.
- ``default_space_id``: optional fallback space ID used by space-related
  scripts when no CLI argument or ``MATRIX_SPACE_ID`` env var is given.

Stdlib only.
"""

from __future__ import annotations

import json
import os
import sys
from pathlib import Path


def get_config_path() -> Path:
    """Return the Matrix configuration file path.

    Respects ``XDG_CONFIG_HOME`` and falls back to ``~/.config``.  Same
    resolution as the matrix-communication skill so a single config file
    is shared by both.
    """
    xdg_config = os.environ.get("XDG_CONFIG_HOME") or (Path.home() / ".config")
    return Path(xdg_config) / "matrix" / "config.json"


def load_config(require_admin: bool = True) -> dict:
    """Load Matrix config from ``$XDG_CONFIG_HOME/matrix/config.json``.

    Args:
        require_admin: If True (the default), require an admin token to be
            present (either as ``admin_token`` or ``access_token``).

    Returns:
        Parsed config dict.

    Exits with a helpful message if the config is missing or incomplete.
    """
    config_path = get_config_path()
    if not config_path.exists():
        print(f"Error: Config file not found: {config_path}", file=sys.stderr)
        print("Create it with at least:", file=sys.stderr)
        example = {
            "homeserver": "https://matrix.example.com",
            "admin_token": "syt_...",
        }
        print(json.dumps(example, indent=2), file=sys.stderr)
        sys.exit(1)

    with open(config_path) as f:
        config = json.load(f)

    if "homeserver" not in config:
        print("Error: config missing required field: homeserver", file=sys.stderr)
        sys.exit(1)

    if require_admin and not (config.get("admin_token") or config.get("access_token")):
        print(
            "Error: config missing admin token. Set 'admin_token' (preferred) "
            "or 'access_token' to a token belonging to a Synapse server admin.",
            file=sys.stderr,
        )
        sys.exit(1)

    return config

"""Human-readable byte-size formatter.

Decimal units by default (``kB``, ``MB``, ``GB`` …); pass ``binary=True``
for IEC units (``KiB``, ``MiB`` …).  Stdlib only.
"""

from __future__ import annotations

_BYTE_UNITS = ["B", "kB", "MB", "GB", "TB", "PB", "EB", "ZB", "YB"]
_BIBYTE_UNITS = ["B", "KiB", "MiB", "GiB", "TiB", "PiB", "EiB", "ZiB", "YiB"]


def pretty_bytes(number: float, binary: bool = False, space: bool = True) -> str:
    """Format ``number`` of bytes with a human-readable unit.

    Examples
    --------
    >>> pretty_bytes(0)
    '0 B'
    >>> pretty_bytes(1500)
    '1.50 kB'
    >>> pretty_bytes(1500, binary=True)
    '1.46 KiB'
    """
    if number != number or number in (float("inf"), float("-inf")):
        raise ValueError(f"Expected a finite number, got: {number!r}")

    units = _BIBYTE_UNITS if binary else _BYTE_UNITS
    sep = " " if space else ""

    sign = "-" if number < 0 else ""
    n = abs(number)

    if n < 1:
        return f"{sign}{n:.0f}{sep}{units[0]}"

    base = 1024 if binary else 1000
    exponent = 0
    while n >= base and exponent < len(units) - 1:
        n /= base
        exponent += 1

    if n >= 100:
        formatted = f"{n:.0f}"
    elif n >= 10:
        formatted = f"{n:.1f}"
    else:
        formatted = f"{n:.2f}"

    return f"{sign}{formatted}{sep}{units[exponent]}"

"""Room health-check rules.

Each room is rated against a set of policies and earns one of three levels:

- ``SUCCESS``: rule satisfied
- ``WARNING``: minor / advisory issue
- ``FAIL``:    serious / blocking issue

Both English and German phrasings are provided.  The "is the room reachable
from one of our spaces" check requires a list of *home* space IDs to be
passed in by the caller — the skill itself ships no homeserver-specific data.

Stdlib only.
"""

from __future__ import annotations

from enum import IntEnum
from typing import Optional

from _lib.condensing import Room


class RoomRating(IntEnum):
    SUCCESS = 0
    WARNING = 1
    FAIL = 2


def rating_emoji(rating: RoomRating) -> str:
    return {RoomRating.SUCCESS: "✅", RoomRating.WARNING: "⚠️", RoomRating.FAIL: "❌"}[
        rating
    ]


def format_rating(item: tuple[RoomRating, str]) -> str:
    rating, message = item
    return f"{rating_emoji(rating)} {message}"


_T = {
    "room": {
        "public": {
            "en": "Is a public room",
            "de": "Ist ein öffentlicher Raum",
        },
        "encryption": {
            "yes": {"en": "Encrypted", "de": "Verschlüsselt"},
            "no": {"en": "Not encrypted", "de": "Nicht verschlüsselt"},
        },
        "joinable-from-our-spaces": {
            "yes": {
                "en": "Joinable from our spaces",
                "de": "Von unseren Spaces aus beitretbar",
            },
            "no": {
                "en": "Not joinable from our spaces",
                "de": "Nicht von unseren Spaces aus beitretbar",
            },
        },
        "in-one-of-our-spaces": {
            "yes": {"en": "In one of our spaces", "de": "In einem unserer Spaces"},
            "no": {
                "en": "Not in one of our spaces",
                "de": "Nicht in einem unserer Spaces",
            },
            "predecessor-was": {
                "en": "Predecessor was in one of our spaces",
                "de": "Vorgänger war in einem unserer Spaces",
            },
        },
    },
    "space": {
        "public": {
            "en": "Is a public space",
            "de": "Ist ein öffentlicher Space",
        },
        "our": {
            "yes": {"en": "One of our spaces", "de": "Einer unserer Spaces"},
            "no": {"en": "Not one of our spaces", "de": "Nicht einer unserer Spaces"},
        },
    },
}


def rate_room(
    r: Room,
    rooms: dict[str, Room],
    home_space_ids: Optional[list[str]] = None,
    language: str = "en",
) -> tuple[RoomRating, list[tuple[RoomRating, str]]]:
    """Run health checks on a single room.

    Args:
        r: The room to rate.
        rooms: All rooms in the snapshot (used to look up parent spaces and
            predecessors).
        home_space_ids: List of "your" space IDs.  Rooms reachable from
            (or members of) one of these are treated as in-space.  When
            ``None`` or empty, the in-space checks degrade to "no opinion".
        language: ``"en"`` or ``"de"``.

    Returns:
        A tuple ``(overall, messages)``: the worst rating across all checks,
        and the per-check ``(rating, message)`` list.
    """
    if language not in ("en", "de"):
        language = "en"
    home = set(home_space_ids or [])

    messages: list[tuple[RoomRating, str]] = []

    def add(rating: RoomRating, message: str) -> None:
        messages.append((rating, message))

    if r.is_space:
        if r.join_policy == "public":
            add(RoomRating.FAIL, _T["space"]["public"][language])

        if isinstance(r.join_policy, tuple) and r.join_policy[0] == "in-space":
            allow_ids = r.join_policy[1]
            if home:
                if not any(rid in home for rid in allow_ids):
                    add(
                        RoomRating.FAIL,
                        _T["room"]["joinable-from-our-spaces"]["no"][language],
                    )
                else:
                    add(
                        RoomRating.SUCCESS,
                        _T["room"]["joinable-from-our-spaces"]["yes"][language],
                    )

        if home:
            if r.id not in home:
                add(RoomRating.WARNING, _T["space"]["our"]["no"][language])
            else:
                add(RoomRating.SUCCESS, _T["space"]["our"]["yes"][language])
    else:
        if r.join_policy == "public":
            add(RoomRating.FAIL, _T["room"]["public"][language])

        if (
            r.version > 9
            and isinstance(r.join_policy, tuple)
            and r.join_policy[0] == "in-space"
        ):
            allow_ids = r.join_policy[1]
            if home:
                if not any(rid in home for rid in allow_ids):
                    add(
                        RoomRating.FAIL,
                        _T["room"]["joinable-from-our-spaces"]["no"][language],
                    )
                else:
                    add(
                        RoomRating.SUCCESS,
                        _T["room"]["joinable-from-our-spaces"]["yes"][language],
                    )

        if home:
            if not _is_in_one_of(r, rooms, home):
                add(
                    RoomRating.WARNING,
                    _T["room"]["in-one-of-our-spaces"]["no"][language],
                )
                if r.predecessor:
                    pred = rooms.get(r.predecessor)
                    if pred and _is_in_one_of(pred, rooms, home):
                        add(
                            RoomRating.FAIL,
                            _T["room"]["in-one-of-our-spaces"]["predecessor-was"][
                                language
                            ],
                        )
            else:
                add(
                    RoomRating.SUCCESS,
                    _T["room"]["in-one-of-our-spaces"]["yes"][language],
                )

        if not r.is_encrypted:
            add(RoomRating.WARNING, _T["room"]["encryption"]["no"][language])
        else:
            add(RoomRating.SUCCESS, _T["room"]["encryption"]["yes"][language])

    overall = max((rating for rating, _ in messages), default=RoomRating.SUCCESS)
    return overall, messages


def _is_in_one_of(r: Room, rooms: dict[str, Room], home: set[str]) -> bool:
    """True iff *some* parent space of ``r`` is in ``home``."""
    for s in rooms.values():
        if r.id in s.space_children and s.id in home:
            return True
    return False

#!/usr/bin/env python3
"""Link a room into a space.

Sends an ``m.space.child`` state event on the space with the room as the
state key.  The admin token must belong to a user that has permission to
send state events in the space (typically a space admin).

Usage:
    synapse-add-to-space.py <ROOM_ID> [SPACE_ID]

Falls back to ``$MATRIX_SPACE_ID`` and then to ``default_space_id`` from
the config when ``SPACE_ID`` is omitted.
"""

from __future__ import annotations

import argparse
import json
import os
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from _lib import client_request, load_config, quote


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    parser.add_argument("room_id")
    parser.add_argument("space_id", nargs="?", default=None)
    parser.add_argument(
        "--via",
        action="append",
        default=None,
        help="Server name to include in 'via'. Repeatable; defaults to the "
        "homeserver hostname.",
    )
    parser.add_argument("--suggested", action="store_true")
    args = parser.parse_args()

    config = load_config()
    space_id = (
        args.space_id
        or os.environ.get("MATRIX_SPACE_ID")
        or config.get("default_space_id")
    )
    if not space_id:
        print(
            "Error: SPACE_ID required (positional, $MATRIX_SPACE_ID, or "
            "'default_space_id' in config).",
            file=sys.stderr,
        )
        return 2

    via = args.via
    if not via:
        from urllib.parse import urlparse

        host = urlparse(config["homeserver"]).hostname
        via = [host] if host else []

    body = {"via": via, "suggested": args.suggested}
    endpoint = f"/rooms/{quote(space_id)}/state/m.space.child/{quote(args.room_id)}"
    result = client_request(config, "PUT", endpoint, body)
    print(json.dumps(result, indent=2))
    return 0 if "error" not in result else 1


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""Print the largest rooms on the homeserver by Synapse-estimated DB size.

Calls ``GET /_synapse/admin/v1/statistics/database/rooms`` and looks up the
display name for each result.  Limit defaults to 10.

Usage:
    synapse-biggest-rooms.py [--limit 10]
"""

from __future__ import annotations

import argparse
import os
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from _lib import admin_request, load_config, pretty_bytes


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    parser.add_argument("-n", "--limit", type=int, default=10)
    args = parser.parse_args()

    config = load_config()
    result = admin_request(config, "GET", "/v1/statistics/database/rooms")
    if "error" in result:
        print(f"Error: {result['error']}", file=sys.stderr)
        return 1

    rooms = result.get("rooms", []) or []
    rooms.sort(key=lambda r: r.get("estimated_size", 0), reverse=True)
    rooms = rooms[: args.limit]

    rows = []
    for r in rooms:
        info = admin_request(config, "GET", f"/v1/rooms/{r['room_id']}")
        name = info.get("name") if isinstance(info, dict) else None
        rows.append(
            {
                "room_id": r["room_id"],
                "name": name,
                "size": pretty_bytes(r.get("estimated_size", 0)),
            }
        )

    width_id = max((len(row["room_id"]) for row in rows), default=8)
    width_size = max((len(row["size"]) for row in rows), default=4)
    print(f"{'room_id'.ljust(width_id)}  {'size'.rjust(width_size)}  name")
    for row in rows:
        name = row["name"] or "-"
        print(
            f"{row['room_id'].ljust(width_id)}  {row['size'].rjust(width_size)}  {name}"
        )
    return 0


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""Deactivate a Matrix user.

Calls ``POST /_synapse/admin/v1/deactivate/{user_id}``.  The user is
unable to log in again afterwards and joined-rooms membership is severed.
**Not reversible** without direct database intervention.

Prints the user's profile and joined-rooms list before and after, so the
console transcript serves as an audit trail.

Usage:
    synapse-deactivate-user.py <USER_ID> [--erase] [--yes]
"""

from __future__ import annotations

import argparse
import json
import os
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from _lib import admin_request, bold, load_config, quote, red, yellow


def _user_info(config: dict, user_id: str) -> dict:
    return admin_request(config, "GET", f"/v2/users/{quote(user_id)}")


def _joined_rooms(config: dict, user_id: str) -> list[str]:
    res = admin_request(config, "GET", f"/v2/users/{quote(user_id)}/joined_rooms")
    return res.get("joined_rooms", []) if isinstance(res, dict) else []


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    parser.add_argument("user_id")
    parser.add_argument(
        "--erase",
        action="store_true",
        help="GDPR erase: also remove the user's messages from rooms.",
    )
    parser.add_argument("-y", "--yes", action="store_true", help="Skip confirmation.")
    args = parser.parse_args()

    config = load_config()

    print(bold("Before:"))
    print(json.dumps(_user_info(config, args.user_id), indent=2))
    print(json.dumps(_joined_rooms(config, args.user_id), indent=2))

    if not args.yes:
        if sys.stdin.isatty():
            print(
                bold(
                    red(f"⚠ About to deactivate {args.user_id}. This cannot be undone.")
                )
            )
            try:
                answer = input("Type 'YES' to continue: ").strip()
            except EOFError:
                answer = ""
            if answer != "YES":
                print(yellow("Aborted."))
                return 1
        else:
            print(
                bold(red("Refusing to run non-interactively without --yes.")),
                file=sys.stderr,
            )
            return 2

    body = {"erase": True} if args.erase else None
    res = admin_request(config, "POST", f"/v1/deactivate/{quote(args.user_id)}", body)
    print(bold("Deactivate response:"))
    print(json.dumps(res, indent=2))

    print(bold("After:"))
    print(json.dumps(_user_info(config, args.user_id), indent=2))
    print(json.dumps(_joined_rooms(config, args.user_id), indent=2))

    return 0 if "error" not in res else 1


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""Snapshot every visible room on a Synapse homeserver.

Pages through ``GET /_synapse/admin/v1/rooms`` and, for each room, fetches
its full state via ``/v1/rooms/{room_id}/state``.  Writes the result as a
JSON list of ``{"room": …, "states": […]}`` entries — the format consumed
by the other scripts (rating, graph, member listings).

Apply an optional server-suffix filter via ``--server`` (e.g.
``--server :example.com``) or the ``room_filter`` config field.  Without a
filter every visible room is included.

Usage:
    synapse-fetch-rooms.py [--output rooms.json] [--server :example.com]
"""

from __future__ import annotations

import argparse
import json
import os
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from _lib import admin_request, load_config, quote


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    parser.add_argument(
        "-o", "--output", default="rooms.json", help="Output path (default rooms.json)"
    )
    parser.add_argument(
        "--server",
        default=None,
        help="Optional server suffix to filter on (e.g. ':example.com'). "
        "Defaults to the 'room_filter' config field.",
    )
    args = parser.parse_args()

    config = load_config()
    suffix = args.server if args.server is not None else config.get("room_filter", "")

    from_token: int | str = 0
    rooms: list[dict] = []

    while True:
        # `order_by=name` is a documented Synapse query parameter; the
        # original tool used a bare `&sort` flag that produced an
        # empty-valued query parameter (parsed inconsistently by some
        # proxies).  See:
        # https://element-hq.github.io/synapse/latest/admin_api/rooms.html
        result = admin_request(
            config,
            "GET",
            f"/v1/rooms?from={quote(str(from_token))}&order_by=name",
        )
        if "error" in result:
            print(f"Error fetching rooms: {result['error']}", file=sys.stderr)
            return 1

        page = result.get("rooms", []) or []
        if suffix:
            page = [r for r in page if r.get("room_id", "").endswith(suffix)]

        for room in page:
            state_res = admin_request(
                config, "GET", f"/v1/rooms/{room['room_id']}/state"
            )
            if "error" in state_res:
                print(
                    f"Warning: failed to fetch state for {room['room_id']}: "
                    f"{state_res['error']}",
                    file=sys.stderr,
                )
                continue
            states = sorted(
                state_res.get("state", []) or [],
                key=lambda s: s.get("origin_server_ts", 0),
            )
            rooms.append({"room": room, "states": states})

        next_batch = result.get("next_batch")
        if next_batch is None:
            break
        from_token = next_batch

    with open(args.output, "w") as f:
        json.dump(rooms, f, indent=2)

    print(f"Wrote {len(rooms)} rooms to {args.output}")
    return 0


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""Render a snapshot of rooms as a Graphviz graph.

Reads ``rooms.json`` (produced by ``synapse-fetch-rooms.py``), writes
``rooms.dot`` (Graphviz source) and — if the ``dot`` binary is available —
``rooms.svg``.

Each room is a node coloured by its rating (green/orange/red).  Spaces are
drawn with a green/blue gradient when healthy.  Edges go from child rooms
to their parent space.

Usage:
    synapse-graph.py [--input rooms.json] [--dot rooms.dot] [--svg rooms.svg] \\
        [--language en|de] [--space '!ID:server' …] [--no-svg]
"""

from __future__ import annotations

import argparse
import json
import os
import shutil
import subprocess
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from _lib import (
    Room,
    RoomRating,
    condense,
    format_rating,
    load_config,
    rate_room,
)


_COLORS = {
    "blue": "#2222ff",
    "bright_blue": "#7777ff",
    "red": "#ff2222",
    "bright_red": "#ff7777",
    "green": "#22cc22",
    "bright_green": "#77ff77",
    "gray": "#222222",
    "orange": "#ff8800",
    "bright_orange": "#ffaa22",
}


def _enc(value: str) -> str:
    """Escape a string for use inside a Graphviz DOT double-quoted literal.

    DOT treats backslash + character as an escape sequence (``\\n``,
    ``\\l``, ``\\r``, ``\\"``).  We escape backslashes first, then quotes,
    then newlines/CR — order matters so the backslash escape doesn't
    double-escape itself.
    """
    if value is None:
        return ""
    return (
        value.replace("\\", "\\\\")
        .replace('"', '\\"')
        .replace("\n", "\\n")
        .replace("\r", "\\r")
    )


def _join_policy_text(r: Room, rooms: dict[str, Room]) -> str:
    if isinstance(r.join_policy, tuple) and r.join_policy[0] == "in-space":
        names = [rooms[rid].name for rid in r.join_policy[1] if rid in rooms]
        return "members of " + ", ".join(names) if names else "members of (unknown)"
    if isinstance(r.join_policy, str):
        return r.join_policy
    return "unknown"


def _icon(r: Room) -> str:
    if r.join_policy == "public":
        return "🌐"
    if r.join_policy == "invite-only":
        return "🔒"
    return "🚀"


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    parser.add_argument("-i", "--input", default="rooms.json")
    parser.add_argument("--dot", default="rooms.dot")
    parser.add_argument("--svg", default="rooms.svg")
    parser.add_argument("-l", "--language", default=None, choices=["en", "de"])
    parser.add_argument(
        "-s",
        "--space",
        action="append",
        default=None,
        help="Space ID treated as 'home'. Repeatable.",
    )
    parser.add_argument(
        "--no-svg", action="store_true", help="Skip the dot → SVG step."
    )
    args = parser.parse_args()

    config = load_config(require_admin=False)
    language = args.language or os.environ.get("LANGUAGE", "en")
    if language not in ("en", "de"):
        language = "en"

    home_spaces: list[str] = list(args.space or [])
    if not home_spaces:
        cfg_home = config.get("home_space_ids") or []
        if isinstance(cfg_home, list):
            home_spaces.extend(cfg_home)
        if config.get("default_space_id"):
            home_spaces.append(config["default_space_id"])

    with open(args.input) as f:
        data = json.load(f)
    rooms = condense(data)

    lines: list[str] = ["digraph G {", '  graph [rankdir="LR"];']
    for r in rooms.values():
        if r.is_replaced:
            continue
        score, violations = rate_room(r, rooms, home_spaces, language)

        if score == RoomRating.SUCCESS:
            color = _COLORS["green"]
            fill = (
                f"{_COLORS['bright_green']}:{_COLORS['bright_blue']}"
                if r.is_space
                else _COLORS["bright_green"]
            )
        elif score == RoomRating.WARNING:
            color = _COLORS["orange"]
            fill = _COLORS["bright_orange"]
        else:
            color = _COLORS["red"]
            fill = _COLORS["bright_red"]

        creator = (r.creator or "unknown").split(":")[0]
        label = f"{_icon(r)} {_enc(r.name)}\\n(v{r.version}, {_join_policy_text(r, rooms)}, by {_enc(creator)})"
        tooltip = "\\n".join(format_rating(m) for m in violations)

        lines.append(
            f'  "{_enc(r.id)}" [label="{label}", style=filled, '
            f'color="{color}", fontcolor="black", fillcolor="{fill}", '
            f'tooltip="{_enc(tooltip)}", shape=rectangle, penwidth=3];'
        )
        for child_id in r.space_children:
            child = rooms.get(child_id)
            if not child or child.is_replaced:
                continue
            lines.append(
                f'  "{_enc(child_id)}" -> "{_enc(r.id)}" [label="space child"];'
            )

    lines.append("}")
    with open(args.dot, "w") as f:
        f.write("\n".join(lines) + "\n")

    if args.no_svg:
        print(f"Wrote {args.dot}")
        return 0

    if shutil.which("dot") is None:
        print(
            f"Wrote {args.dot}. Install Graphviz ('dot') to render the SVG, or pass --no-svg.",
            file=sys.stderr,
        )
        return 0

    try:
        subprocess.run(
            ["dot", "-Tsvg", args.dot, "-o", args.svg],
            check=True,
        )
    except subprocess.CalledProcessError as e:
        print(f"dot failed: {e}", file=sys.stderr)
        return 1

    print(f"Wrote {args.dot} and {args.svg}")
    return 0


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""Force-join a user into a room using the Synapse Admin API.

Calls ``POST /_synapse/admin/v1/join/{room_id}`` with the target user.
At least one user that the admin can impersonate (or the admin themselves)
must already be a member of the room — otherwise Synapse refuses.

Usage:
    synapse-join-room.py <ROOM_ID> [USER_ID]

Falls back to ``$MATRIX_USER_ID`` when ``USER_ID`` is omitted.
"""

from __future__ import annotations

import argparse
import json
import os
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from _lib import admin_request, load_config


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    parser.add_argument("room_id")
    parser.add_argument("user_id", nargs="?", default=None)
    args = parser.parse_args()

    config = load_config()
    user_id = args.user_id or os.environ.get("MATRIX_USER_ID")
    if not user_id:
        print(
            "Error: USER_ID required (positional argument or $MATRIX_USER_ID).",
            file=sys.stderr,
        )
        return 2

    result = admin_request(
        config, "POST", f"/v1/join/{args.room_id}", {"user_id": user_id}
    )
    print(json.dumps(result, indent=2))
    return 0 if "error" not in result else 1


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""Promote a user to power-level 100 in a room.

Calls ``POST /_synapse/admin/v1/rooms/{room_id}/make_room_admin``.
Synapse only lets this succeed while at least one existing room admin is
still a member.  If the original owner has left, the room is unrecoverable
through this tool.

Usage:
    synapse-make-admin.py <ROOM_ID> [USER_ID]

Falls back to ``$MATRIX_USER_ID`` when ``USER_ID`` is omitted.
"""

from __future__ import annotations

import argparse
import json
import os
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from _lib import admin_request, load_config


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    parser.add_argument("room_id")
    parser.add_argument("user_id", nargs="?", default=None)
    args = parser.parse_args()

    config = load_config()
    user_id = args.user_id or os.environ.get("MATRIX_USER_ID")
    if not user_id:
        print(
            "Error: USER_ID required (positional argument or $MATRIX_USER_ID).",
            file=sys.stderr,
        )
        return 2

    result = admin_request(
        config,
        "POST",
        f"/v1/rooms/{args.room_id}/make_room_admin",
        {"user_id": user_id},
    )
    print(json.dumps(result, indent=2))
    return 0 if "error" not in result else 1


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""Harden a room: link it into a space, restrict joins, enable encryption.

Multi-step operation:

1. Add the room to the target space (``m.space.child`` state event), unless
   already present.
2. Force-join the calling user.
3. If the user has no explicit power level, temporarily promote them to
   admin (PL 100).
4. If the room is currently public *and* its room-version supports
   restricted joins (>9), switch ``m.room.join_rules`` to ``restricted`` so
   only members of the parent space can join.
5. Enable Megolm encryption (``m.room.encryption``) if not already enabled.
6. Restore the user's original power level — runs in a ``finally`` block
   and on SIGINT/SIGTERM, so a crash or Ctrl-C does not leave the user
   with elevated permissions.

WARNING: enabling encryption is irreversible.  Restricted joins remove
discoverability for users outside the space.

Usage:
    synapse-migrate-room.py <ROOM_ID> [USER_ID] [SPACE_ID]

Falls back to ``$MATRIX_USER_ID`` / ``$MATRIX_SPACE_ID`` /
``default_space_id``.
"""

from __future__ import annotations

import argparse
import os
import signal
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from _lib import (
    admin_request,
    bold,
    client_request,
    green,
    load_config,
    quote,
    red,
    yellow,
)


def _state(config: dict, room_id: str) -> list[dict]:
    res = client_request(config, "GET", f"/rooms/{quote(room_id)}/state")
    if isinstance(res, list):
        return res
    return res.get("state", []) or []


def _put_state(
    config: dict,
    room_id: str,
    event_type: str,
    state_key: str,
    body: dict,
) -> dict:
    endpoint = f"/rooms/{quote(room_id)}/state/{quote(event_type)}/{quote(state_key)}"
    return client_request(config, "PUT", endpoint, body)


def _join(config: dict, room_id: str) -> dict:
    return client_request(config, "POST", f"/rooms/{quote(room_id)}/join", {})


def _restore_power_level(
    config: dict,
    room_id: str,
    user_id: str,
    pl_content: dict,
    previous_level: int | None,
) -> None:
    """Restore ``user_id``'s entry in the room's ``m.room.power_levels``.

    If ``previous_level`` is ``None`` the user originally had no explicit
    entry and ``users[user_id]`` is removed (so the user falls back to
    ``users_default``).  Otherwise the entry is set to ``previous_level``.

    Best-effort: errors are reported but never raised.
    """
    if not pl_content:
        return
    users = dict(pl_content.get("users") or {})
    if previous_level is None:
        if user_id not in users:
            return
        print(bold(yellow(f"⚠ Removing explicit PL entry for {user_id}")))
        users.pop(user_id, None)
    else:
        if users.get(user_id) == previous_level:
            return
        print(bold(yellow(f"⚠ Restoring {user_id} to power level {previous_level}")))
        users[user_id] = previous_level
    new_pl = dict(pl_content)
    new_pl["users"] = users
    res = _put_state(config, room_id, "m.room.power_levels", "", new_pl)
    if "error" in res:
        print(red(f"✗ Failed to restore power level: {res['error']}"), file=sys.stderr)
    else:
        print(bold(green("✓ Power levels restored")))


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    parser.add_argument("room_id")
    parser.add_argument("user_id", nargs="?", default=None)
    parser.add_argument("space_id", nargs="?", default=None)
    args = parser.parse_args()

    config = load_config()
    user_id = args.user_id or os.environ.get("MATRIX_USER_ID")
    space_id = (
        args.space_id
        or os.environ.get("MATRIX_SPACE_ID")
        or config.get("default_space_id")
    )

    if not user_id:
        print("Error: USER_ID required.", file=sys.stderr)
        return 2
    if not space_id:
        print(
            "Error: SPACE_ID required (positional, $MATRIX_SPACE_ID, or "
            "'default_space_id' in config).",
            file=sys.stderr,
        )
        return 2

    from urllib.parse import urlparse

    via_host = urlparse(config["homeserver"]).hostname
    via = [via_host] if via_host else []

    # 1. Add to space if not already a child.
    space_state = _state(config, space_id)
    already_child = any(
        s.get("type") == "m.space.child"
        and s.get("state_key") == args.room_id
        and (s.get("content") or {})
        for s in space_state
    )
    if already_child:
        print(bold(green("✓ Room already in space")))
    else:
        print(bold(yellow("⚠ Adding room to space")))
        res = _put_state(
            config,
            space_id,
            "m.space.child",
            args.room_id,
            {"via": via, "suggested": False},
        )
        if "error" in res:
            print(red(f"✗ Failed to add to space: {res['error']}"), file=sys.stderr)
            return 1
        print(bold(green("✓ Room added to space")))

    # 2. Join.
    join_res = _join(config, args.room_id)
    if "error" in join_res:
        print(red(f"⚠ Join: {join_res['error']}"), file=sys.stderr)
    else:
        print(bold(green("✓ Joined")))

    # Inspect target room (needed for power-level snapshot before any change).
    room_info = admin_request(config, "GET", f"/v1/rooms/{args.room_id}")
    if "error" in room_info:
        print(red(f"✗ {room_info['error']}"), file=sys.stderr)
        return 1
    room_state = _state(config, args.room_id)

    pl_event = next(
        (s for s in room_state if s.get("type") == "m.room.power_levels"), None
    )
    pl_content = (pl_event or {}).get("content", {}) or {}
    users = dict(pl_content.get("users") or {})
    users_default = pl_content.get("users_default", 0)

    # `previous_level` retains its `None` if the user had no explicit entry.
    previous_level: int | None = users.get(user_id)
    had_explicit_entry = user_id in users
    promoted = False

    def _try_promote() -> bool:
        res = admin_request(
            config,
            "POST",
            f"/v1/rooms/{args.room_id}/make_room_admin",
            {"user_id": user_id},
        )
        if "error" in res:
            print(red(f"✗ Promote failed: {res['error']}"), file=sys.stderr)
            return False
        return True

    if not had_explicit_entry:
        print(
            bold(
                yellow(
                    f"⚠ {user_id} has no explicit power level (default {users_default}); "
                    f"will restore after operations"
                )
            )
        )
        promoted = _try_promote()
    elif previous_level == 100:
        print(bold(green("✓ User already has maximum power level")))
    else:
        print(
            bold(
                green(
                    f"✓ User has power level {previous_level}; will restore after operations"
                )
            )
        )
        promoted = _try_promote()

    if not promoted and not had_explicit_entry and previous_level is None:
        # We never managed to elevate; subsequent state writes will likely
        # fail.  Bail out cleanly without touching power levels.
        print(
            red("✗ Could not elevate user; skipping the rest of the pipeline."),
            file=sys.stderr,
        )
        return 1

    # Install a SIGINT/SIGTERM handler so Ctrl-C still restores the power
    # level before exiting.
    def _signal_handler(signum, _frame):
        if promoted:
            _restore_power_level(
                config, args.room_id, user_id, pl_content, previous_level
            )
        sys.exit(128 + signum)

    if promoted:
        signal.signal(signal.SIGINT, _signal_handler)
        signal.signal(signal.SIGTERM, _signal_handler)

    try:
        # 3. Restrict joins if currently public.
        join_rules = (room_info.get("join_rules") or "").lower()
        try:
            version = int(room_info.get("version") or 0)
        except (TypeError, ValueError):
            version = 0

        if join_rules == "public":
            if version > 9:
                print(bold(yellow("⚠ Setting join rules to restricted")))
                res = _put_state(
                    config,
                    args.room_id,
                    "m.room.join_rules",
                    "",
                    {
                        "join_rule": "restricted",
                        "allow": [{"type": "m.room_membership", "room_id": space_id}],
                    },
                )
                if "error" in res:
                    print(red(f"✗ Failed: {res['error']}"), file=sys.stderr)
                else:
                    print(bold(green("✓ Join rules set to restricted")))
            else:
                print(
                    bold(
                        red(
                            f"✗ Room version {version} is too low for restricted joins; "
                            "skipping"
                        )
                    )
                )
        else:
            print(bold(green("✓ Join rules already restricted (or not public)")))

        # 4. Encrypt if not already.
        if not room_info.get("encryption"):
            print(bold(yellow("⚠ Enabling encryption")))
            res = _put_state(
                config,
                args.room_id,
                "m.room.encryption",
                "",
                {"algorithm": "m.megolm.v1.aes-sha2"},
            )
            if "error" in res:
                print(red(f"✗ Failed: {res['error']}"), file=sys.stderr)
            else:
                print(bold(green("✓ Encryption enabled")))
        else:
            print(bold(green("✓ Encryption already enabled")))
    finally:
        # 5. Always restore the previous power level if we promoted the user,
        # even on exception.
        if promoted:
            _restore_power_level(
                config, args.room_id, user_id, pl_content, previous_level
            )

    return 0


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""Rate every room in a snapshot and print the failing ones.

Reads ``rooms.json`` (produced by ``synapse-fetch-rooms.py``), runs the
health-check rules from ``_lib/rating.py`` against every non-replaced room,
and prints two formatted lists of rooms with at least one ``FAIL`` finding:

1. Jira wiki markup
2. Markdown

Use ``--language de`` for German phrasing.  Pass ``--space`` one or more
times (or set ``default_space_id``/``home_space_ids`` in the config) to
enable the "is in one of our spaces" checks; without it those checks are
skipped.

Usage:
    synapse-rate-rooms.py [--input rooms.json] [--language en|de] \\
        [--space '!ID:server' …]
"""

from __future__ import annotations

import argparse
import json
import os
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from _lib import (
    RoomRating,
    condense,
    format_rating,
    load_config,
    rate_room,
)


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    parser.add_argument("-i", "--input", default="rooms.json")
    parser.add_argument("-l", "--language", default=None, choices=["en", "de"])
    parser.add_argument(
        "-s",
        "--space",
        action="append",
        default=None,
        help="Space ID treated as 'home'. Repeatable.",
    )
    args = parser.parse_args()

    config = load_config(require_admin=False)

    language = args.language or os.environ.get("LANGUAGE", "en")
    if language not in ("en", "de"):
        language = "en"

    home_spaces: list[str] = list(args.space or [])
    if not home_spaces:
        cfg_home = config.get("home_space_ids") or []
        if isinstance(cfg_home, list):
            home_spaces.extend(cfg_home)
        if config.get("default_space_id"):
            home_spaces.append(config["default_space_id"])

    with open(args.input) as f:
        data = json.load(f)
    rooms = condense(data)

    flagged = []
    for r in rooms.values():
        if r.is_replaced:
            continue
        overall, messages = rate_room(r, rooms, home_spaces, language)
        ratings = [m for m in messages if m[0] != RoomRating.SUCCESS]
        if any(rating == RoomRating.FAIL for rating, _ in ratings):
            flagged.append((r, ratings))

    flagged.sort(key=lambda pair: pair[0].name.strip().lower())

    def render(formatter):
        return "\n".join(formatter(r, ratings) for r, ratings in flagged)

    def jira(r, ratings):
        head = f"{{task}}{r.name.strip()} (v{r.version}){{task}}"
        body = "\n".join(f"** {format_rating(m)}" for m in ratings)
        return f"{head}\n{body}" if body else head

    def markdown(r, ratings):
        head = f"- {r.name.strip()} (v{r.version})"
        body = "\n".join(f"  {format_rating(m)}" for m in ratings)
        return f"{head}\n{body}" if body else head

    print(render(jira))
    print("\n\n")
    print(render(markdown))
    return 0


if __name__ == "__main__":
    sys.exit(main())