Simplify the alias deletion logic as an application service. (#13093)

This commit is contained in:
Quentin Gliech 2022-06-17 13:19:22 +02:00 committed by GitHub
parent 5ef05c70c3
commit 73af10f419
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 48 additions and 22 deletions

1
changelog.d/13093.misc Normal file
View File

@ -0,0 +1 @@
Simplify the alias deletion logic as an application service.

View File

@ -17,13 +17,7 @@ from typing import TYPE_CHECKING, Tuple
from twisted.web.server import Request from twisted.web.server import Request
from synapse.api.errors import ( from synapse.api.errors import AuthError, Codes, NotFoundError, SynapseError
AuthError,
Codes,
InvalidClientCredentialsError,
NotFoundError,
SynapseError,
)
from synapse.http.server import HttpServer from synapse.http.server import HttpServer
from synapse.http.servlet import RestServlet, parse_json_object_from_request from synapse.http.servlet import RestServlet, parse_json_object_from_request
from synapse.http.site import SynapseRequest from synapse.http.site import SynapseRequest
@ -96,30 +90,27 @@ class ClientDirectoryServer(RestServlet):
self, request: SynapseRequest, room_alias: str self, request: SynapseRequest, room_alias: str
) -> Tuple[int, JsonDict]: ) -> Tuple[int, JsonDict]:
room_alias_obj = RoomAlias.from_string(room_alias) room_alias_obj = RoomAlias.from_string(room_alias)
requester = await self.auth.get_user_by_req(request)
try: if requester.app_service:
service = self.auth.get_appservice_by_req(request)
await self.directory_handler.delete_appservice_association( await self.directory_handler.delete_appservice_association(
service, room_alias_obj requester.app_service, room_alias_obj
) )
logger.info( logger.info(
"Application service at %s deleted alias %s", "Application service at %s deleted alias %s",
service.url, requester.app_service.url,
room_alias_obj.to_string(), room_alias_obj.to_string(),
) )
return 200, {}
except InvalidClientCredentialsError:
# fallback to default user behaviour if they aren't an AS
pass
requester = await self.auth.get_user_by_req(request) else:
user = requester.user await self.directory_handler.delete_association(requester, room_alias_obj)
await self.directory_handler.delete_association(requester, room_alias_obj) logger.info(
"User %s deleted alias %s",
logger.info( requester.user.to_string(),
"User %s deleted alias %s", user.to_string(), room_alias_obj.to_string() room_alias_obj.to_string(),
) )
return 200, {} return 200, {}

View File

@ -16,6 +16,7 @@ from http import HTTPStatus
from twisted.test.proto_helpers import MemoryReactor from twisted.test.proto_helpers import MemoryReactor
from synapse.appservice import ApplicationService
from synapse.rest import admin from synapse.rest import admin
from synapse.rest.client import directory, login, room from synapse.rest.client import directory, login, room
from synapse.server import HomeServer from synapse.server import HomeServer
@ -129,6 +130,39 @@ class DirectoryTestCase(unittest.HomeserverTestCase):
) )
self.assertEqual(channel.code, HTTPStatus.OK, channel.result) self.assertEqual(channel.code, HTTPStatus.OK, channel.result)
def test_deleting_alias_via_directory_appservice(self) -> None:
user_id = "@as:test"
as_token = "i_am_an_app_service"
appservice = ApplicationService(
as_token,
id="1234",
namespaces={"aliases": [{"regex": "#asns-*", "exclusive": True}]},
sender=user_id,
)
self.hs.get_datastores().main.services_cache.append(appservice)
# Add an alias for the room, as the appservice
alias = RoomAlias(f"asns-{random_string(5)}", self.hs.hostname).to_string()
data = {"room_id": self.room_id}
request_data = json.dumps(data)
channel = self.make_request(
"PUT",
f"/_matrix/client/r0/directory/room/{alias}",
request_data,
access_token=as_token,
)
self.assertEqual(channel.code, HTTPStatus.OK, channel.result)
# Then try to remove the alias, as the appservice
channel = self.make_request(
"DELETE",
f"/_matrix/client/r0/directory/room/{alias}",
access_token=as_token,
)
self.assertEqual(channel.code, HTTPStatus.OK, channel.result)
def test_deleting_nonexistant_alias(self) -> None: def test_deleting_nonexistant_alias(self) -> None:
# Check that no alias exists # Check that no alias exists
alias = "#potato:test" alias = "#potato:test"