diff options
Diffstat (limited to 'synapse')
| -rw-r--r-- | synapse/api/urls.py | 1 | ||||
| -rwxr-xr-x | synapse/app/homeserver.py | 7 | ||||
| -rw-r--r-- | synapse/appservice/__init__.py | 137 | ||||
| -rw-r--r-- | synapse/appservice/api.py | 110 | ||||
| -rw-r--r-- | synapse/handlers/__init__.py | 2 | ||||
| -rw-r--r-- | synapse/handlers/appservice.py | 195 | ||||
| -rw-r--r-- | synapse/handlers/directory.py | 7 | ||||
| -rw-r--r-- | synapse/handlers/login.py | 33 | ||||
| -rw-r--r-- | synapse/handlers/register.py | 56 | ||||
| -rw-r--r-- | synapse/http/client.py | 66 | ||||
| -rw-r--r-- | synapse/notifier.py | 6 | ||||
| -rw-r--r-- | synapse/rest/appservice/__init__.py | 14 | ||||
| -rw-r--r-- | synapse/rest/appservice/v1/__init__.py | 29 | ||||
| -rw-r--r-- | synapse/rest/appservice/v1/base.py | 48 | ||||
| -rw-r--r-- | synapse/rest/appservice/v1/register.py | 121 | ||||
| -rw-r--r-- | synapse/server.py | 1 | ||||
| -rw-r--r-- | synapse/storage/__init__.py | 3 | ||||
| -rw-r--r-- | synapse/storage/appservice.py | 244 | ||||
| -rw-r--r-- | synapse/storage/schema/application_services.sql | 33 |
19 files changed, 1065 insertions, 48 deletions
diff --git a/synapse/api/urls.py b/synapse/api/urls.py index 693c0efda6..9485719332 100644 --- a/synapse/api/urls.py +++ b/synapse/api/urls.py @@ -22,3 +22,4 @@ WEB_CLIENT_PREFIX = "/_matrix/client" CONTENT_REPO_PREFIX = "/_matrix/content" SERVER_KEY_PREFIX = "/_matrix/key/v1" MEDIA_PREFIX = "/_matrix/media/v1" +APP_SERVICE_PREFIX = "/_matrix/appservice/v1" diff --git a/synapse/app/homeserver.py b/synapse/app/homeserver.py index f20ccfb5b6..a9397de5b2 100755 --- a/synapse/app/homeserver.py +++ b/synapse/app/homeserver.py @@ -26,13 +26,14 @@ from twisted.web.resource import Resource from twisted.web.static import File from twisted.web.server import Site from synapse.http.server import JsonResource, RootRedirect +from synapse.rest.appservice.v1 import AppServiceRestResource from synapse.rest.media.v0.content_repository import ContentRepoResource from synapse.rest.media.v1.media_repository import MediaRepositoryResource from synapse.http.server_key_resource import LocalKey from synapse.http.matrixfederationclient import MatrixFederationHttpClient from synapse.api.urls import ( CLIENT_PREFIX, FEDERATION_PREFIX, WEB_CLIENT_PREFIX, CONTENT_REPO_PREFIX, - SERVER_KEY_PREFIX, MEDIA_PREFIX, CLIENT_V2_ALPHA_PREFIX, + SERVER_KEY_PREFIX, MEDIA_PREFIX, CLIENT_V2_ALPHA_PREFIX, APP_SERVICE_PREFIX ) from synapse.config.homeserver import HomeServerConfig from synapse.crypto import context_factory @@ -69,6 +70,9 @@ class SynapseHomeServer(HomeServer): def build_resource_for_federation(self): return JsonResource() + def build_resource_for_app_services(self): + return AppServiceRestResource(self) + def build_resource_for_web_client(self): syweb_path = os.path.dirname(syweb.__file__) webclient_path = os.path.join(syweb_path, "webclient") @@ -114,6 +118,7 @@ class SynapseHomeServer(HomeServer): (CONTENT_REPO_PREFIX, self.get_resource_for_content_repo()), (SERVER_KEY_PREFIX, self.get_resource_for_server_key()), (MEDIA_PREFIX, self.get_resource_for_media_repository()), + (APP_SERVICE_PREFIX, self.get_resource_for_app_services()), ] if web_client: logger.info("Adding the web client.") diff --git a/synapse/appservice/__init__.py b/synapse/appservice/__init__.py new file mode 100644 index 0000000000..46d46a5a48 --- /dev/null +++ b/synapse/appservice/__init__.py @@ -0,0 +1,137 @@ +# -*- coding: utf-8 -*- +# Copyright 2015 OpenMarket Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +from synapse.api.constants import EventTypes + +import logging +import re + +logger = logging.getLogger(__name__) + + +class ApplicationService(object): + """Defines an application service. This definition is mostly what is + provided to the /register AS API. + + Provides methods to check if this service is "interested" in events. + """ + NS_USERS = "users" + NS_ALIASES = "aliases" + NS_ROOMS = "rooms" + # The ordering here is important as it is used to map database values (which + # are stored as ints representing the position in this list) to namespace + # values. + NS_LIST = [NS_USERS, NS_ALIASES, NS_ROOMS] + + def __init__(self, token, url=None, namespaces=None, hs_token=None, + txn_id=None): + self.token = token + self.url = url + self.hs_token = hs_token + self.namespaces = self._check_namespaces(namespaces) + self.txn_id = txn_id + + def _check_namespaces(self, namespaces): + # Sanity check that it is of the form: + # { + # users: ["regex",...], + # aliases: ["regex",...], + # rooms: ["regex",...], + # } + if not namespaces: + return None + + for ns in ApplicationService.NS_LIST: + if type(namespaces[ns]) != list: + raise ValueError("Bad namespace value for '%s'", ns) + for regex in namespaces[ns]: + if not isinstance(regex, basestring): + raise ValueError("Expected string regex for ns '%s'", ns) + return namespaces + + def _matches_regex(self, test_string, namespace_key): + if not isinstance(test_string, basestring): + logger.error( + "Expected a string to test regex against, but got %s", + test_string + ) + return False + + for regex in self.namespaces[namespace_key]: + if re.match(regex, test_string): + return True + return False + + def _matches_user(self, event): + if (hasattr(event, "sender") and + self.is_interested_in_user(event.sender)): + return True + # also check m.room.member state key + if (hasattr(event, "type") and event.type == EventTypes.Member + and hasattr(event, "state_key") + and self.is_interested_in_user(event.state_key)): + return True + return False + + def _matches_room_id(self, event): + if hasattr(event, "room_id"): + return self.is_interested_in_room(event.room_id) + return False + + def _matches_aliases(self, event, alias_list): + for alias in alias_list: + if self.is_interested_in_alias(alias): + return True + return False + + def is_interested(self, event, restrict_to=None, aliases_for_event=None): + """Check if this service is interested in this event. + + Args: + event(Event): The event to check. + restrict_to(str): The namespace to restrict regex tests to. + aliases_for_event(list): A list of all the known room aliases for + this event. + Returns: + bool: True if this service would like to know about this event. + """ + if aliases_for_event is None: + aliases_for_event = [] + if restrict_to and restrict_to not in ApplicationService.NS_LIST: + # this is a programming error, so fail early and raise a general + # exception + raise Exception("Unexpected restrict_to value: %s". restrict_to) + + if not restrict_to: + return (self._matches_user(event) + or self._matches_aliases(event, aliases_for_event) + or self._matches_room_id(event)) + elif restrict_to == ApplicationService.NS_ALIASES: + return self._matches_aliases(event, aliases_for_event) + elif restrict_to == ApplicationService.NS_ROOMS: + return self._matches_room_id(event) + elif restrict_to == ApplicationService.NS_USERS: + return self._matches_user(event) + + def is_interested_in_user(self, user_id): + return self._matches_regex(user_id, ApplicationService.NS_USERS) + + def is_interested_in_alias(self, alias): + return self._matches_regex(alias, ApplicationService.NS_ALIASES) + + def is_interested_in_room(self, room_id): + return self._matches_regex(room_id, ApplicationService.NS_ROOMS) + + def __str__(self): + return "ApplicationService: %s" % (self.__dict__,) diff --git a/synapse/appservice/api.py b/synapse/appservice/api.py new file mode 100644 index 0000000000..9cce4e0973 --- /dev/null +++ b/synapse/appservice/api.py @@ -0,0 +1,110 @@ +# -*- coding: utf-8 -*- +# Copyright 2015 OpenMarket Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +from twisted.internet import defer + +from synapse.api.errors import CodeMessageException +from synapse.http.client import SimpleHttpClient +from synapse.events.utils import serialize_event + +import logging +import urllib + +logger = logging.getLogger(__name__) + + +class ApplicationServiceApi(SimpleHttpClient): + """This class manages HS -> AS communications, including querying and + pushing. + """ + + def __init__(self, hs): + super(ApplicationServiceApi, self).__init__(hs) + self.clock = hs.get_clock() + + @defer.inlineCallbacks + def query_user(self, service, user_id): + uri = service.url + ("/users/%s" % urllib.quote(user_id)) + response = None + try: + response = yield self.get_json(uri, { + "access_token": service.hs_token + }) + if response: # just an empty json object + defer.returnValue(True) + except CodeMessageException as e: + if e.code == 404: + defer.returnValue(False) + return + logger.warning("query_user to %s received %s", uri, e.code) + except Exception as ex: + logger.warning("query_user to %s threw exception %s", uri, ex) + defer.returnValue(False) + + @defer.inlineCallbacks + def query_alias(self, service, alias): + uri = service.url + ("/rooms/%s" % urllib.quote(alias)) + response = None + try: + response = yield self.get_json(uri, { + "access_token": service.hs_token + }) + if response: # just an empty json object + defer.returnValue(True) + except CodeMessageException as e: + if e.code == 404: + defer.returnValue(False) + return + logger.warning("query_alias to %s received %s", uri, e.code) + except Exception as ex: + logger.warning("query_alias to %s threw exception %s", uri, ex) + defer.returnValue(False) + + + @defer.inlineCallbacks + def push_bulk(self, service, events): + events = self._serialize(events) + + uri = service.url + ("/transactions/%s" % + urllib.quote(str(0))) # TODO txn_ids + response = None + try: + response = yield self.put_json( + uri, + { + "events": events + }, + { + "access_token": service.hs_token + }) + if response: # just an empty json object + # TODO: Mark txn as sent successfully + defer.returnValue(True) + except CodeMessageException as e: + logger.warning("push_bulk to %s received %s", uri, e.code) + except Exception as ex: + logger.warning("push_bulk to %s threw exception %s", uri, ex) + defer.returnValue(False) + + @defer.inlineCallbacks + def push(self, service, event): + response = yield self.push_bulk(service, [event]) + defer.returnValue(response) + + def _serialize(self, events): + time_now = self.clock.time_msec() + return [ + serialize_event(e, time_now, as_client_event=True) for e in events + ] + diff --git a/synapse/handlers/__init__.py b/synapse/handlers/__init__.py index a32eab9316..b31518bf62 100644 --- a/synapse/handlers/__init__.py +++ b/synapse/handlers/__init__.py @@ -26,6 +26,7 @@ from .presence import PresenceHandler from .directory import DirectoryHandler from .typing import TypingNotificationHandler from .admin import AdminHandler +from .appservice import ApplicationServicesHandler from .sync import SyncHandler @@ -52,4 +53,5 @@ class Handlers(object): self.directory_handler = DirectoryHandler(hs) self.typing_notification_handler = TypingNotificationHandler(hs) self.admin_handler = AdminHandler(hs) + self.appservice_handler = ApplicationServicesHandler(hs) self.sync_handler = SyncHandler(hs) diff --git a/synapse/handlers/appservice.py b/synapse/handlers/appservice.py new file mode 100644 index 0000000000..8d0cdd528c --- /dev/null +++ b/synapse/handlers/appservice.py @@ -0,0 +1,195 @@ +# -*- coding: utf-8 -*- +# Copyright 2015 OpenMarket Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from twisted.internet import defer + +from synapse.api.constants import EventTypes +from synapse.api.errors import Codes, StoreError, SynapseError +from synapse.appservice import ApplicationService +from synapse.appservice.api import ApplicationServiceApi +from synapse.types import UserID +import synapse.util.stringutils as stringutils + +import logging + +logger = logging.getLogger(__name__) + + +# NB: Purposefully not inheriting BaseHandler since that contains way too much +# setup code which this handler does not need or use. This makes testing a lot +# easier. +class ApplicationServicesHandler(object): + + def __init__(self, hs): + self.store = hs.get_datastore() + self.hs = hs + self.appservice_api = ApplicationServiceApi(hs) + + @defer.inlineCallbacks + def register(self, app_service): + logger.info("Register -> %s", app_service) + # check the token is recognised + try: + stored_service = yield self.store.get_app_service_by_token( + app_service.token + ) + if not stored_service: + raise StoreError(404, "Application service not found") + except StoreError: + raise SynapseError( + 403, "Unrecognised application services token. " + "Consult the home server admin.", + errcode=Codes.FORBIDDEN + ) + logger.info("Updating application service info...") + app_service.hs_token = self._generate_hs_token() + yield self.store.update_app_service(app_service) + defer.returnValue(app_service) + + @defer.inlineCallbacks + def unregister(self, token): + logger.info("Unregister as_token=%s", token) + yield self.store.unregister_app_service(token) + + @defer.inlineCallbacks + def notify_interested_services(self, event): + """Notifies (pushes) all application services interested in this event. + + Pushing is done asynchronously, so this method won't block for any + prolonged length of time. + + Args: + event(Event): The event to push out to interested services. + """ + # Gather interested services + services = yield self._get_services_for_event(event) + if len(services) == 0: + return # no services need notifying + + # Do we know this user exists? If not, poke the user query API for + # all services which match that user regex. This needs to block as these + # user queries need to be made BEFORE pushing the event. + yield self._check_user_exists(event.sender) + if event.type == EventTypes.Member: + yield self._check_user_exists(event.state_key) + + # Fork off pushes to these services - XXX First cut, best effort + for service in services: + self.appservice_api.push(service, event) + + @defer.inlineCallbacks + def query_user_exists(self, user_id): + """Check if any application service knows this user_id exists. + + Args: + user_id(str): The user to query if they exist on any AS. + Returns: + True if this user exists on at least one application service. + """ + user_query_services = yield self._get_services_for_user( + user_id=user_id + ) + for user_service in user_query_services: + is_known_user = yield self.appservice_api.query_user( + user_service, user_id + ) + if is_known_user: + defer.returnValue(True) + defer.returnValue(False) + + @defer.inlineCallbacks + def query_room_alias_exists(self, room_alias): + """Check if an application service knows this room alias exists. + + Args: + room_alias(RoomAlias): The room alias to query. + Returns: + namedtuple: with keys "room_id" and "servers" or None if no + association can be found. + """ + room_alias = room_alias.to_string() + alias_query_services = yield self._get_services_for_event( + event=None, + restrict_to=ApplicationService.NS_ALIASES, + alias_list=[room_alias] + ) + for alias_service in alias_query_services: + is_known_alias = yield self.appservice_api.query_alias( + alias_service, room_alias + ) + if is_known_alias: + # the alias exists now so don't query more ASes. + result = yield self.store.get_association_from_room_alias( + room_alias + ) + defer.returnValue(result) + + @defer.inlineCallbacks + def _get_services_for_event(self, event, restrict_to="", alias_list=None): + """Retrieve a list of application services interested in this event. + + Args: + event(Event): The event to check. Can be None if alias_list is not. + restrict_to(str): The namespace to restrict regex tests to. + alias_list: A list of aliases to get services for. If None, this + list is obtained from the database. + Returns: + list<ApplicationService>: A list of services interested in this + event based on the service regex. + """ + # We need to know the aliases associated with this event.room_id, if any + if not alias_list: + alias_list = yield self.store.get_aliases_for_room(event.room_id) + + services = yield self.store.get_app_services() + interested_list = [ + s for s in services if ( + s.is_interested(event, restrict_to, alias_list) + ) + ] + defer.returnValue(interested_list) + + @defer.inlineCallbacks + def _get_services_for_user(self, user_id): + services = yield self.store.get_app_services() + interested_list = [ + s for s in services if ( + s.is_interested_in_user(user_id) + ) + ] + defer.returnValue(interested_list) + + @defer.inlineCallbacks + def _is_unknown_user(self, user_id): + user = UserID.from_string(user_id) + if not self.hs.is_mine(user): + # we don't know if they are unknown or not since it isn't one of our + # users. We can't poke ASes. + defer.returnValue(False) + return + + user_info = yield self.store.get_user_by_id(user_id) + defer.returnValue(len(user_info) == 0) + + @defer.inlineCallbacks + def _check_user_exists(self, user_id): + unknown_user = yield self._is_unknown_user(user_id) + if unknown_user: + exists = yield self.query_user_exists(user_id) + defer.returnValue(exists) + defer.returnValue(True) + + def _generate_hs_token(self): + return stringutils.random_string(24) diff --git a/synapse/handlers/directory.py b/synapse/handlers/directory.py index 7b60921040..24ea3573d3 100644 --- a/synapse/handlers/directory.py +++ b/synapse/handlers/directory.py @@ -84,6 +84,13 @@ class DirectoryHandler(BaseHandler): if result: room_id = result.room_id servers = result.servers + else: + # Query AS to see if it exists + as_handler = self.hs.get_handlers().appservice_handler + result = yield as_handler.query_room_alias_exists(room_alias) + if result: + room_id = result.room_id + servers = result.servers else: try: result = yield self.federation.make_query( diff --git a/synapse/handlers/login.py b/synapse/handlers/login.py index d297d71c03..7447800460 100644 --- a/synapse/handlers/login.py +++ b/synapse/handlers/login.py @@ -16,12 +16,13 @@ from twisted.internet import defer from ._base import BaseHandler -from synapse.api.errors import LoginError, Codes +from synapse.api.errors import LoginError, Codes, CodeMessageException from synapse.http.client import SimpleHttpClient from synapse.util.emailutils import EmailException import synapse.util.emailutils as emailutils import bcrypt +import json import logging logger = logging.getLogger(__name__) @@ -96,16 +97,20 @@ class LoginHandler(BaseHandler): @defer.inlineCallbacks def _query_email(self, email): - httpCli = SimpleHttpClient(self.hs) - data = yield httpCli.get_json( - # TODO FIXME This should be configurable. - # XXX: ID servers need to use HTTPS - "http://%s%s" % ( - "matrix.org:8090", "/_matrix/identity/api/v1/lookup" - ), - { - 'medium': 'email', - 'address': email - } - ) - defer.returnValue(data) + http_client = SimpleHttpClient(self.hs) + try: + data = yield http_client.get_json( + # TODO FIXME This should be configurable. + # XXX: ID servers need to use HTTPS + "http://%s%s" % ( + "matrix.org:8090", "/_matrix/identity/api/v1/lookup" + ), + { + 'medium': 'email', + 'address': email + } + ) + defer.returnValue(data) + except CodeMessageException as e: + data = json.loads(e.msg) + defer.returnValue(data) diff --git a/synapse/handlers/register.py b/synapse/handlers/register.py index 66a89c10b2..08cd5fd720 100644 --- a/synapse/handlers/register.py +++ b/synapse/handlers/register.py @@ -18,7 +18,7 @@ from twisted.internet import defer from synapse.types import UserID from synapse.api.errors import ( - SynapseError, RegistrationError, InvalidCaptchaError + SynapseError, RegistrationError, InvalidCaptchaError, CodeMessageException ) from ._base import BaseHandler import synapse.util.stringutils as stringutils @@ -28,6 +28,7 @@ from synapse.http.client import CaptchaServerHttpClient import base64 import bcrypt +import json import logging logger = logging.getLogger(__name__) @@ -161,21 +162,26 @@ class RegistrationHandler(BaseHandler): def _threepid_from_creds(self, creds): # TODO: get this from the homeserver rather than creating a new one for # each request - httpCli = SimpleHttpClient(self.hs) + http_client = SimpleHttpClient(self.hs) # XXX: make this configurable! trustedIdServers = ['matrix.org:8090', 'matrix.org'] if not creds['idServer'] in trustedIdServers: logger.warn('%s is not a trusted ID server: rejecting 3pid ' + 'credentials', creds['idServer']) defer.returnValue(None) - data = yield httpCli.get_json( - # XXX: This should be HTTPS - "http://%s%s" % ( - creds['idServer'], - "/_matrix/identity/api/v1/3pid/getValidated3pid" - ), - {'sid': creds['sid'], 'clientSecret': creds['clientSecret']} - ) + + data = {} + try: + data = yield http_client.get_json( + # XXX: This should be HTTPS + "http://%s%s" % ( + creds['idServer'], + "/_matrix/identity/api/v1/3pid/getValidated3pid" + ), + {'sid': creds['sid'], 'clientSecret': creds['clientSecret']} + ) + except CodeMessageException as e: + data = json.loads(e.msg) if 'medium' in data: defer.returnValue(data) @@ -185,19 +191,23 @@ class RegistrationHandler(BaseHandler): def _bind_threepid(self, creds, mxid): yield logger.debug("binding threepid") - httpCli = SimpleHttpClient(self.hs) - data = yield httpCli.post_urlencoded_get_json( - # XXX: Change when ID servers are all HTTPS - "http://%s%s" % ( - creds['idServer'], "/_matrix/identity/api/v1/3pid/bind" - ), - { - 'sid': creds['sid'], - 'clientSecret': creds['clientSecret'], - 'mxid': mxid, - } - ) - logger.debug("bound threepid") + http_client = SimpleHttpClient(self.hs) + data = None + try: + data = yield http_client.post_urlencoded_get_json( + # XXX: Change when ID servers are all HTTPS + "http://%s%s" % ( + creds['idServer'], "/_matrix/identity/api/v1/3pid/bind" + ), + { + 'sid': creds['sid'], + 'clientSecret': creds['clientSecret'], + 'mxid': mxid, + } + ) + logger.debug("bound threepid") + except CodeMessageException as e: + data = json.loads(e.msg) defer.returnValue(data) @defer.inlineCallbacks diff --git a/synapse/http/client.py b/synapse/http/client.py index 198f575cfa..575510637e 100644 --- a/synapse/http/client.py +++ b/synapse/http/client.py @@ -13,7 +13,7 @@ # See the License for the specific language governing permissions and # limitations under the License. - +from synapse.api.errors import CodeMessageException from synapse.http.agent_name import AGENT_NAME from twisted.internet import defer, reactor from twisted.web.client import ( @@ -83,7 +83,7 @@ class SimpleHttpClient(object): @defer.inlineCallbacks def get_json(self, uri, args={}): - """ Get's some json from the given host and path + """ Gets some json from the given URI. Args: uri (str): The URI to request, not including query parameters @@ -91,15 +91,12 @@ class SimpleHttpClient(object): None. **Note**: The value of each key is assumed to be an iterable and *not* a string. - Returns: - Deferred: Succeeds when we get *any* HTTP response. - - The result of the deferred is a tuple of `(code, response)`, - where `response` is a dict representing the decoded JSON body. + Deferred: Succeeds when we get *any* 2xx HTTP response, with the + HTTP body as JSON. + Raises: + On a non-2xx HTTP response. """ - - yield if len(args): query_bytes = urllib.urlencode(args, True) uri = "%s?%s" % (uri, query_bytes) @@ -114,7 +111,56 @@ class SimpleHttpClient(object): body = yield readBody(response) - defer.returnValue(json.loads(body)) + if 200 <= response.code < 300: + defer.returnValue(json.loads(body)) + else: + # NB: This is explicitly not json.loads(body)'d because the contract + # of CodeMessageException is a *string* message. Callers can always + # load it into JSON if they want. + raise CodeMessageException(response.code, body) + + @defer.inlineCallbacks + def put_json(self, uri, json_body, args={}): + """ Puts some json to the given URI. + + Args: + uri (str): The URI to request, not including query parameters + json_body (dict): The JSON to put in the HTTP body, + args (dict): A dictionary used to create query strings, defaults to + None. + **Note**: The value of each key is assumed to be an iterable + and *not* a string. + Returns: + Deferred: Succeeds when we get *any* 2xx HTTP response, with the + HTTP body as JSON. + Raises: + On a non-2xx HTTP response. + """ + if len(args): + query_bytes = urllib.urlencode(args, True) + uri = "%s?%s" % (uri, query_bytes) + + json_str = json.dumps(json_body) + + response = yield self.agent.request( + "PUT", + uri.encode("ascii"), + headers=Headers({ + b"User-Agent": [AGENT_NAME], + "Content-Type": ["application/json"] + }), + bodyProducer=FileBodyProducer(StringIO(json_str)) + ) + + body = yield readBody(response) + + if 200 <= response.code < 300: + defer.returnValue(json.loads(body)) + else: + # NB: This is explicitly not json.loads(body)'d because the contract + # of CodeMessageException is a *string* message. Callers can always + # load it into JSON if they want. + raise CodeMessageException(response.code, body) class CaptchaServerHttpClient(SimpleHttpClient): diff --git a/synapse/notifier.py b/synapse/notifier.py index e3b6ead620..c7f75ab801 100644 --- a/synapse/notifier.py +++ b/synapse/notifier.py @@ -99,6 +99,12 @@ class Notifier(object): `extra_users` param. """ yield run_on_reactor() + + # poke any interested application service. + self.hs.get_handlers().appservice_handler.notify_interested_services( + event + ) + room_id = event.room_id room_source = self.event_sources.sources["room"] diff --git a/synapse/rest/appservice/__init__.py b/synapse/rest/appservice/__init__.py new file mode 100644 index 0000000000..1a84d94cd9 --- /dev/null +++ b/synapse/rest/appservice/__init__.py @@ -0,0 +1,14 @@ +# -*- coding: utf-8 -*- +# Copyright 2015 OpenMarket Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. diff --git a/synapse/rest/appservice/v1/__init__.py b/synapse/rest/appservice/v1/__init__.py new file mode 100644 index 0000000000..bf243b6180 --- /dev/null +++ b/synapse/rest/appservice/v1/__init__.py @@ -0,0 +1,29 @@ +# -*- coding: utf-8 -*- +# Copyright 2015 OpenMarket Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +from . import register + +from synapse.http.server import JsonResource + + +class AppServiceRestResource(JsonResource): + """A resource for version 1 of the matrix application service API.""" + + def __init__(self, hs): + JsonResource.__init__(self) + self.register_servlets(self, hs) + + @staticmethod + def register_servlets(appservice_resource, hs): + register.register_servlets(hs, appservice_resource) \ No newline at end of file diff --git a/synapse/rest/appservice/v1/base.py b/synapse/rest/appservice/v1/base.py new file mode 100644 index 0000000000..65d5bcf9be --- /dev/null +++ b/synapse/rest/appservice/v1/base.py @@ -0,0 +1,48 @@ +# -*- coding: utf-8 -*- +# Copyright 2015 OpenMarket Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""This module contains base REST classes for constructing client v1 servlets. +""" + +from synapse.http.servlet import RestServlet +from synapse.api.urls import APP_SERVICE_PREFIX +import re + +import logging + + +logger = logging.getLogger(__name__) + + +def as_path_pattern(path_regex): + """Creates a regex compiled appservice path with the correct path + prefix. + + Args: + path_regex (str): The regex string to match. This should NOT have a ^ + as this will be prefixed. + Returns: + SRE_Pattern + """ + return re.compile("^" + APP_SERVICE_PREFIX + path_regex) + + +class AppServiceRestServlet(RestServlet): + """A base Synapse REST Servlet for the application services version 1 API. + """ + + def __init__(self, hs): + self.hs = hs + self.handler = hs.get_handlers().appservice_handler diff --git a/synapse/rest/appservice/v1/register.py b/synapse/rest/appservice/v1/register.py new file mode 100644 index 0000000000..d3d5aef220 --- /dev/null +++ b/synapse/rest/appservice/v1/register.py @@ -0,0 +1,121 @@ +# -*- coding: utf-8 -*- +# Copyright 2015 OpenMarket Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""This module contains REST servlets to do with registration: /register""" +from twisted.internet import defer + +from base import AppServiceRestServlet, as_path_pattern +from synapse.api.errors import CodeMessageException, SynapseError +from synapse.storage.appservice import ApplicationService + +import json +import logging + +logger = logging.getLogger(__name__) + + +class RegisterRestServlet(AppServiceRestServlet): + """Handles AS registration with the home server. + """ + + PATTERN = as_path_pattern("/register$") + + @defer.inlineCallbacks + def on_POST(self, request): + params = _parse_json(request) + + # sanity check required params + try: + as_token = params["as_token"] + as_url = params["url"] + if (not isinstance(as_token, basestring) or + not isinstance(as_url, basestring)): + raise ValueError + except (KeyError, ValueError): + raise SynapseError( + 400, "Missed required keys: as_token(str) / url(str)." + ) + + namespaces = { + "users": [], + "rooms": [], + "aliases": [] + } + + if "namespaces" in params: + self._parse_namespace(namespaces, params["namespaces"], "users") + self._parse_namespace(namespaces, params["namespaces"], "rooms") + self._parse_namespace(namespaces, params["namespaces"], "aliases") + + app_service = ApplicationService(as_token, as_url, namespaces) + + app_service = yield self.handler.register(app_service) + hs_token = app_service.hs_token + + defer.returnValue((200, { + "hs_token": hs_token + })) + + def _parse_namespace(self, target_ns, origin_ns, ns): + if ns not in target_ns or ns not in origin_ns: + return # nothing to parse / map through to. + + possible_regex_list = origin_ns[ns] + if not type(possible_regex_list) == list: + raise SynapseError(400, "Namespace %s isn't an array." % ns) + + for regex in possible_regex_list: + if not isinstance(regex, basestring): + raise SynapseError( + 400, "Regex '%s' isn't a string in namespace %s" % + (regex, ns) + ) + + target_ns[ns] = origin_ns[ns] + + +class UnregisterRestServlet(AppServiceRestServlet): + """Handles AS registration with the home server. + """ + + PATTERN = as_path_pattern("/unregister$") + + def on_POST(self, request): + params = _parse_json(request) + try: + as_token = params["as_token"] + if not isinstance(as_token, basestring): + raise ValueError + except (KeyError, ValueError): + raise SynapseError(400, "Missing required key: as_token(str)") + + yield self.handler.unregister(as_token) + + raise CodeMessageException(500, "Not implemented") + + +def _parse_json(request): + try: + content = json.loads(request.content.read()) + if type(content) != dict: + raise SynapseError(400, "Content must be a JSON object.") + return content + except ValueError: + raise SynapseError(400, "Content not JSON.") + + +def register_servlets(hs, http_server): + RegisterRestServlet(hs).register(http_server) + UnregisterRestServlet(hs).register(http_server) diff --git a/synapse/server.py b/synapse/server.py index 23bdad0c7c..ba2b2593f1 100644 --- a/synapse/server.py +++ b/synapse/server.py @@ -77,6 +77,7 @@ class BaseHomeServer(object): 'resource_for_content_repo', 'resource_for_server_key', 'resource_for_media_repository', + 'resource_for_app_services', 'event_sources', 'ratelimiter', 'keyring', diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index a63c59a8a2..2225be96be 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -18,6 +18,7 @@ from twisted.internet import defer from synapse.util.logutils import log_function from synapse.api.constants import EventTypes +from .appservice import ApplicationServiceStore from .directory import DirectoryStore from .feedback import FeedbackStore from .presence import PresenceStore @@ -66,6 +67,7 @@ SCHEMAS = [ "event_signatures", "pusher", "media_repository", + "application_services", "filtering", "rejections", ] @@ -87,6 +89,7 @@ class DataStore(RoomMemberStore, RoomStore, RegistrationStore, StreamStore, ProfileStore, FeedbackStore, PresenceStore, TransactionStore, DirectoryStore, KeyStore, StateStore, SignatureStore, + ApplicationServiceStore, EventFederationStore, MediaRepositoryStore, RejectionsStore, diff --git a/synapse/storage/appservice.py b/synapse/storage/appservice.py new file mode 100644 index 0000000000..3c8bf9ad0d --- /dev/null +++ b/synapse/storage/appservice.py @@ -0,0 +1,244 @@ +# -*- coding: utf-8 -*- +# Copyright 2015 OpenMarket Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +import logging +from twisted.internet import defer + +from synapse.api.errors import StoreError +from synapse.appservice import ApplicationService +from ._base import SQLBaseStore + + +logger = logging.getLogger(__name__) + + +class ApplicationServiceCache(object): + """Caches ApplicationServices and provides utility functions on top. + + This class is designed to be invoked on incoming events in order to avoid + hammering the database every time to extract a list of application service + regexes. + """ + + def __init__(self): + self.services = [] + + +class ApplicationServiceStore(SQLBaseStore): + + def __init__(self, hs): + super(ApplicationServiceStore, self).__init__(hs) + self.cache = ApplicationServiceCache() + self.cache_defer = self._populate_cache() + + @defer.inlineCallbacks + def unregister_app_service(self, token): + """Unregisters this service. + + This removes all AS specific regex and the base URL. The token is the + only thing preserved for future registration attempts. + """ + yield self.cache_defer # make sure the cache is ready + yield self.runInteraction( + "unregister_app_service", + self._unregister_app_service_txn, + token, + ) + # update cache TODO: Should this be in the txn? + for service in self.cache.services: + if service.token == token: + service.url = None + service.namespaces = None + service.hs_token = None + + def _unregister_app_service_txn(self, txn, token): + # kill the url to prevent pushes + txn.execute( + "UPDATE application_services SET url=NULL WHERE token=?", + (token,) + ) + + # cleanup regex + as_id = self._get_as_id_txn(txn, token) + if not as_id: + logger.warning( + "unregister_app_service_txn: Failed to find as_id for token=", + token + ) + return False + + txn.execute( + "DELETE FROM application_services_regex WHERE as_id=?", + (as_id,) + ) + return True + + @defer.inlineCallbacks + def update_app_service(self, service): + """Update an application service, clobbering what was previously there. + + Args: + service(ApplicationService): The updated service. + """ + yield self.cache_defer # make sure the cache is ready + + # NB: There is no "insert" since we provide no public-facing API to + # allocate new ASes. It relies on the server admin inserting the AS + # token into the database manually. + + + if not service.token or not service.url: + raise StoreError(400, "Token and url must be specified.") + + if not service.hs_token: + raise StoreError(500, "No HS token") + + yield self.runInteraction( + "update_app_service", + self._update_app_service_txn, + service + ) + + # update cache TODO: Should this be in the txn? + for (index, cache_service) in enumerate(self.cache.services): + if service.token == cache_service.token: + self.cache.services[index] = service + logger.info("Updated: %s", service) + return + # new entry + self.cache.services.append(service) + logger.info("Updated(new): %s", service) + + def _update_app_service_txn(self, txn, service): + as_id = self._get_as_id_txn(txn, service.token) + if not as_id: + logger.warning( + "update_app_service_txn: Failed to find as_id for token=", + service.token + ) + return False + + txn.execute( + "UPDATE application_services SET url=?, hs_token=? WHERE id=?", + (service.url, service.hs_token, as_id,) + ) + # cleanup regex + txn.execute( + "DELETE FROM application_services_regex WHERE as_id=?", + (as_id,) + ) + for (ns_int, ns_str) in enumerate(ApplicationService.NS_LIST): + if ns_str in service.namespaces: + for regex in service.namespaces[ns_str]: + txn.execute( + "INSERT INTO application_services_regex(" + "as_id, namespace, regex) values(?,?,?)", + (as_id, ns_int, regex) + ) + return True + + def _get_as_id_txn(self, txn, token): + cursor = txn.execute( + "SELECT id FROM application_services WHERE token=?", + (token,) + ) + res = cursor.fetchone() + if res: + return res[0] + + @defer.inlineCallbacks + def get_app_services(self): + yield self.cache_defer # make sure the cache is ready + defer.returnValue(self.cache.services) + + @defer.inlineCallbacks + def get_app_service_by_token(self, token, from_cache=True): + """Get the application service with the given token. + + Args: + token (str): The application service token. + from_cache (bool): True to get this service from the cache, False to + check the database. + Raises: + StoreError if there was a problem retrieving this service. + """ + yield self.cache_defer # make sure the cache is ready + + if from_cache: + for service in self.cache.services: + if service.token == token: + defer.returnValue(service) + return + defer.returnValue(None) + + # TODO: The from_cache=False impl + # TODO: This should be JOINed with the application_services_regex table. + + + @defer.inlineCallbacks + def _populate_cache(self): + """Populates the ApplicationServiceCache from the database.""" + sql = ("SELECT * FROM application_services LEFT JOIN " + "application_services_regex ON application_services.id = " + "application_services_regex.as_id") + # SQL results in the form: + # [ + # { + # 'regex': "something", + # 'url': "something", + # 'namespace': enum, + # 'as_id': 0, + # 'token': "something", + # 'hs_token': "otherthing", + # 'id': 0 + # } + # ] + services = {} + results = yield self._execute_and_decode(sql) + for res in results: + as_token = res["token"] + if as_token not in services: + # add the service + services[as_token] = { + "url": res["url"], + "token": as_token, + "hs_token": res["hs_token"], + "namespaces": { + ApplicationService.NS_USERS: [], + ApplicationService.NS_ALIASES: [], + ApplicationService.NS_ROOMS: [] + } + } + # add the namespace regex if one exists + ns_int = res["namespace"] + if ns_int is None: + continue + try: + services[as_token]["namespaces"][ + ApplicationService.NS_LIST[ns_int]].append( + res["regex"] + ) + except IndexError: + logger.error("Bad namespace enum '%s'. %s", ns_int, res) + + # TODO get last successful txn id f.e. service + for service in services.values(): + logger.info("Found application service: %s", service) + self.cache.services.append(ApplicationService( + token=service["token"], + url=service["url"], + namespaces=service["namespaces"], + hs_token=service["hs_token"] + )) + diff --git a/synapse/storage/schema/application_services.sql b/synapse/storage/schema/application_services.sql new file mode 100644 index 0000000000..03b5a10c8a --- /dev/null +++ b/synapse/storage/schema/application_services.sql @@ -0,0 +1,33 @@ +/* Copyright 2015 OpenMarket Ltd + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +CREATE TABLE IF NOT EXISTS application_services( + id INTEGER PRIMARY KEY AUTOINCREMENT, + url TEXT, + token TEXT, + hs_token TEXT, + UNIQUE(token) ON CONFLICT ROLLBACK +); + +CREATE TABLE IF NOT EXISTS application_services_regex( + id INTEGER PRIMARY KEY AUTOINCREMENT, + as_id INTEGER NOT NULL, + namespace INTEGER, /* enum[room_id|room_alias|user_id] */ + regex TEXT, + FOREIGN KEY(as_id) REFERENCES application_services(id) +); + + + |