Admin API implementation

matrix_client/admin_api.py

@@ -0,0 +1,116 @@
+# -*- coding: utf-8 -*-
+from matrix_client.api import MatrixHttpApi
+try:
+    basestring
+except NameError:
+    basestring = str
+
+
+class MatrixHttpAdminApi(MatrixHttpApi):
+    """Extends Matrix API with admin calls.
+
+    Examples:
+        Create a client and send a message::
+
+            matrix = MatrixHttpAdminApi("https://matrix.org", token="foobar")
+            response = admin_api.shutdown_room(
+                "!DgvjtOljKujDBrxyHk:matrix.org",
+                "@admin:matrix.org",
+                room_name="New room",
+                message="Old room closed by admin"
+            )
+    """
+    def purge_history(self, room_id, event_id):
+        """Perform /admin/purge_hostory.
+        Admin api part.
+        Args:
+            room_id (str): Room_id to purge.
+            event_id (str or int): Event_id or ts to purge before.
+        """
+        if isinstance(event_id, basestring):
+            content = {
+                "delete_local_events": True,
+                "purge_up_to_event_id": event_id
+            }
+        else:
+            content = {
+                "delete_local_events": True,
+                "purge_up_to_ts": int(event_id)
+            }
+        return self._send("POST", "/admin/purge_history/%s" % room_id, content)
+
+    def purge_history_status(self, purge_id):
+        """Perform /admin/purge_history_status.
+        Admin api part.
+        Args:
+            purge_id (str): Purge_id to query status.
+        """
+        return self._send("GET", "/admin/purge_history_status/%s" % purge_id)
+
+    def media_in_room(self, room_id, event_id=None):
+        """List remote and local media in room.
+        Args:
+            room_id (str): Room_id to purge.
+        """
+        return self._send("GET", "/admin/room/%s/media" % room_id)
+
+    def whois(self, user_id):
+        """Query server for user information (ip, UA, last seen).
+        Admin api part.
+        Args:
+            user_id (str): user_id to query.
+        """
+        return self._send("GET", "/admin/whois/%s" % user_id)
+
+    def deactivate(self, user_id, erase=False):
+        """Deactivate user account.
+        Admin api part.
+        Args:
+            user_id (str): user_id to deactivate.
+            erase (bool): erase user data. Default no.
+        """
+        content = {
+            "erase": erase
+        }
+        return self._send("POST", "/admin/deactivate/%s" % user_id, content)
+
+    def reset_password(self, user_id, password):
+        """Reset users's password to provided.
+        Admin api part.
+        Args:
+            user_id (str): user_id to deactivate.
+            password (str): password to set.
+        """
+        content = {
+            "new_password": password
+        }
+        return self._send("POST", "/admin/reset_password/%s" % user_id, content)
+
+    def quarantine_media(self, room_id):
+        """Quarantine all media in room so that no one can download it via thi server.
+        Admin api part.
+        Args:
+            room_id (str): room_id to quarantine.
+        """
+        return self._send("POST", "/admin/quarantine_media/%s" % room_id)
+
+    def shutdown_room(self, room_id, new_room_user_id, room_name=False, message=False):
+        """Shuts down a room by removing all local users from the room and blocking
+        all future invites and joins to the room. Any local aliases will be repointed
+        to a new room created by `new_room_user_id` and kicked users will be auto
+        joined to the new room
+        Admin api part.
+        Args:
+            room_id (str): room_id to quarantine.
+            new_room_user_id (str): new room creator user_id.
+            room_name (str): new room name.
+            message (str): information message for new room.
+        """
+        content = {
+            "new_room_user_id": new_room_user_id
+        }
+        if room_name:
+            content["room_name"] = room_name
+        if message:
+            content["message"] = message
+        return self._send("POST", "/admin/shutdown_room/%s" % room_id, content)

test/admin_api_test.py

@@ -0,0 +1,169 @@
+import responses
+import json
+from matrix_client.admin_api import MatrixHttpAdminApi
+
+
+class TestAdminApi:
+    admin_api = MatrixHttpAdminApi("http://example.com")
+    user_id = "@alice:matrix.org"
+    room_id = "!gveUzqBzXPqmwvDaCZ:example.org"
+    event_id = "$153119074937XoqNn::example.org"
+    up_to_ts = 1531190749090
+    purge_id = "dLVEjckmfggyQduS"
+
+    @responses.activate
+    def test_purge_history_eventid(self):
+        purge_history_url = \
+            "http://example.com/_matrix/client/r0/" \
+            "admin/purge_history/%s" % self.room_id
+        responses.add(
+            responses.POST,
+            purge_history_url,
+            body='{"purge_id": "%s"}' % self.purge_id
+        )
+        self.admin_api.purge_history(self.room_id, self.event_id)
+        req = responses.calls[0].request
+        assert req.url == purge_history_url
+        assert req.method == 'POST'
+        j = json.loads(req.body)
+        assert j["delete_local_events"]
+        assert j["purge_up_to_event_id"] == self.event_id
+
+    @responses.activate
+    def test_purge_history_up_to_ts(self):
+        purge_history_url = \
+            "http://example.com/_matrix/client/r0/" \
+            "admin/purge_history/%s" % self.room_id
+        responses.add(
+            responses.POST,
+            purge_history_url,
+            body='{"purge_id": "%s"}' % self.purge_id
+        )
+        self.admin_api.purge_history(self.room_id, self.up_to_ts)
+        req = responses.calls[0].request
+        j = json.loads(req.body)
+        assert j["delete_local_events"]
+        assert j["purge_up_to_ts"] == self.up_to_ts
+
+    @responses.activate
+    def test_purge_history_status(self):
+        purge_history_status_url = \
+            "http://example.com/_matrix/client/r0/" \
+            "admin/purge_history_status/%s" % self.purge_id
+        responses.add(
+            responses.GET,
+            purge_history_status_url,
+            body='{"status": "complete"}'
+        )
+        self.admin_api.purge_history_status(self.purge_id)
+        req = responses.calls[0].request
+        assert req.url == purge_history_status_url
+
+    @responses.activate
+    def test_media_in_room(self):
+        media_url = \
+            "http://example.com/_matrix/client/r0/" \
+            "admin/room/%s/media" % self.room_id
+        responses.add(
+            responses.GET,
+            media_url,
+            body='{"local": ["mxc://example.com/xwvutsrqponmlkjihgfedcba"],'
+            ' "remote": ["mxc://matrix.org/xwtttsrqponmlkjihgfedcba"]}'
+        )
+        resp = self.admin_api.media_in_room(self.room_id)
+        req = responses.calls[0].request
+        assert req.url == media_url
+        assert req.method == 'GET'
+        assert "local" in resp
+        assert "remote" in resp
+
+    @responses.activate
+    def test_whois(self):
+        whois_url = \
+            "http://example.com/_matrix/client/r0/" \
+            "admin/whois/%s" % self.user_id
+        responses.add(
+            responses.GET,
+            whois_url,
+            body='{"user_id": "%s", "devices": {}}' % self.user_id
+        )
+        self.admin_api.whois(self.user_id)
+        req = responses.calls[0].request
+        assert req.url == whois_url
+        assert req.method == 'GET'
+
+    @responses.activate
+    def test_deactivate_no_erase(self):
+        erase_url = \
+            "http://example.com/_matrix/client/r0/" \
+            "admin/deactivate/%s" % self.user_id
+        responses.add(responses.POST, erase_url, body='{}')
+        self.admin_api.deactivate(self.user_id)
+        req = responses.calls[0].request
+        assert req.url == erase_url
+        assert req.method == 'POST'
+
+    @responses.activate
+    def test_deactivate(self):
+        erase_url = \
+            "http://example.com/_matrix/client/r0/" \
+            "admin/deactivate/%s" % self.user_id
+        responses.add(responses.POST, erase_url, body='{}')
+        self.admin_api.deactivate(self.user_id, erase=True)
+        req = responses.calls[0].request
+        assert req.url == erase_url
+        assert req.method == 'POST'
+        j = json.loads(req.body)
+        assert j["erase"]
+
+    @responses.activate
+    def test_reset_password(self):
+        reset_url = \
+            "http://example.com/_matrix/client/r0/" \
+            "admin/reset_password/%s" % self.user_id
+        responses.add(responses.POST, reset_url, body='{}')
+        self.admin_api.reset_password(self.user_id, 'secret')
+        req = responses.calls[0].request
+        assert req.url == reset_url
+        assert req.method == 'POST'
+        j = json.loads(req.body)
+        assert j["new_password"] == 'secret'
+
+    @responses.activate
+    def test_quarantine_media(self):
+        quarantine_media_url = \
+            "http://example.com/_matrix/client/r0/" \
+            "admin/quarantine_media/%s" % self.room_id
+        responses.add(
+            responses.POST,
+            quarantine_media_url,
+            body='{"num_quarantined": 1}'
+        )
+        self.admin_api.quarantine_media(self.room_id)
+        req = responses.calls[0].request
+        assert req.url == quarantine_media_url
+        assert req.method == 'POST'
+
+    @responses.activate
+    def test_shutdown_room(self):
+        shutdown_room_url = \
+            "http://example.com/_matrix/client/r0/" \
+            "admin/shutdown_room/%s" % self.room_id
+        responses.add(
+            responses.POST,
+            shutdown_room_url,
+            body='{"kicked_users": 2, '
+            '"local_aliases": [], '
+            '"new_room_id": "!hepuyalbwtkjapqdhq:example.org"}'
+        )
+        self.admin_api.shutdown_room(
+            self.room_id,
+            self.user_id,
+            room_name="New room",
+            message="Old room closed by admin"
+        )
+        req = responses.calls[0].request
+        assert req.url == shutdown_room_url
+        assert req.method == 'POST'
+        j = json.loads(req.body)
+        assert j["new_room_user_id"] == self.user_id