Mirrors
/
synapse-old
mirror of https://github.com/matrix-org/synapse.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #2466 from matrix-org/erikj/groups_merged 

Initial Group Implementation
This commit is contained in:
Erik Johnston 2017-10-11 13:20:07 +01:00 committed by GitHub
parent dfbf73408c ec954f47fb
commit 535cc49f27
35 changed files with 4793 additions and 66 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
synapse/app/synchrotron.py
View File

@ -40,6 +40,7 @@ from synapse.replication.slave.storage.push_rule import SlavedPushRuleStore
from synapse.replication.slave.storage.receipts import SlavedReceiptsStore
from synapse.replication.slave.storage.registration import SlavedRegistrationStore
from synapse.replication.slave.storage.room import RoomStore
from synapse.replication.slave.storage.groups import SlavedGroupServerStore
from synapse.replication.tcp.client import ReplicationClientHandler
from synapse.rest.client.v1 import events
from synapse.rest.client.v1.initial_sync import InitialSyncRestServlet
@ -69,6 +70,7 @@ class SynchrotronSlavedStore(
SlavedRegistrationStore,
SlavedFilteringStore,
SlavedPresenceStore,
SlavedGroupServerStore,
SlavedDeviceInboxStore,
SlavedDeviceStore,
SlavedClientIpStore,
@ -403,6 +405,10 @@ class SyncReplicationHandler(ReplicationClientHandler):
)
elif stream_name == "presence":
yield self.presence_handler.process_replication_rows(token, rows)
elif stream_name == "receipts":
self.notifier.on_new_event(
"groups_key", token, users=[row.user_id for row in rows],
)
def start(config_options):

368
synapse/federation/transport/client.py
View File

@ -471,3 +471,371 @@ class TransportLayerClient(object):
)
defer.returnValue(content)
@log_function
def get_group_profile(self, destination, group_id, requester_user_id):
"""Get a group profile
"""
path = PREFIX + "/groups/%s/profile" % (group_id,)
return self.client.get_json(
destination=destination,
path=path,
args={"requester_user_id": requester_user_id},
ignore_backoff=True,
)
@log_function
def get_group_summary(self, destination, group_id, requester_user_id):
"""Get a group summary
"""
path = PREFIX + "/groups/%s/summary" % (group_id,)
return self.client.get_json(
destination=destination,
path=path,
args={"requester_user_id": requester_user_id},
ignore_backoff=True,
)
@log_function
def get_rooms_in_group(self, destination, group_id, requester_user_id):
"""Get all rooms in a group
"""
path = PREFIX + "/groups/%s/rooms" % (group_id,)
return self.client.get_json(
destination=destination,
path=path,
args={"requester_user_id": requester_user_id},
ignore_backoff=True,
)
def add_room_to_group(self, destination, group_id, requester_user_id, room_id,
content):
"""Add a room to a group
"""
path = PREFIX + "/groups/%s/room/%s" % (group_id, room_id,)
return self.client.post_json(
destination=destination,
path=path,
args={"requester_user_id": requester_user_id},
data=content,
ignore_backoff=True,
)
def remove_room_from_group(self, destination, group_id, requester_user_id, room_id):
"""Remove a room from a group
"""
path = PREFIX + "/groups/%s/room/%s" % (group_id, room_id,)
return self.client.delete_json(
destination=destination,
path=path,
args={"requester_user_id": requester_user_id},
ignore_backoff=True,
)
@log_function
def get_users_in_group(self, destination, group_id, requester_user_id):
"""Get users in a group
"""
path = PREFIX + "/groups/%s/users" % (group_id,)
return self.client.get_json(
destination=destination,
path=path,
args={"requester_user_id": requester_user_id},
ignore_backoff=True,
)
@log_function
def accept_group_invite(self, destination, group_id, user_id, content):
"""Accept a group invite
"""
path = PREFIX + "/groups/%s/users/%s/accept_invite" % (group_id, user_id)
return self.client.post_json(
destination=destination,
path=path,
data=content,
ignore_backoff=True,
)
@log_function
def invite_to_group(self, destination, group_id, user_id, requester_user_id, content):
"""Invite a user to a group
"""
path = PREFIX + "/groups/%s/users/%s/invite" % (group_id, user_id)
return self.client.post_json(
destination=destination,
path=path,
args=requester_user_id,
data=content,
ignore_backoff=True,
)
@log_function
def invite_to_group_notification(self, destination, group_id, user_id, content):
"""Sent by group server to inform a user's server that they have been
invited.
"""
path = PREFIX + "/groups/local/%s/users/%s/invite" % (group_id, user_id)
return self.client.post_json(
destination=destination,
path=path,
data=content,
ignore_backoff=True,
)
@log_function
def remove_user_from_group(self, destination, group_id, requester_user_id,
user_id, content):
"""Remove a user fron a group
"""
path = PREFIX + "/groups/%s/users/%s/remove" % (group_id, user_id)
return self.client.post_json(
destination=destination,
path=path,
args={"requester_user_id": requester_user_id},
data=content,
ignore_backoff=True,
)
@log_function
def remove_user_from_group_notification(self, destination, group_id, user_id,
content):
"""Sent by group server to inform a user's server that they have been
kicked from the group.
"""
path = PREFIX + "/groups/local/%s/users/%s/remove" % (group_id, user_id)
return self.client.post_json(
destination=destination,
path=path,
data=content,
ignore_backoff=True,
)
@log_function
def renew_group_attestation(self, destination, group_id, user_id, content):
"""Sent by either a group server or a user's server to periodically update
the attestations
"""
path = PREFIX + "/groups/%s/renew_attestation/%s" % (group_id, user_id)
return self.client.post_json(
destination=destination,
path=path,
data=content,
ignore_backoff=True,
)
@log_function
def update_group_summary_room(self, destination, group_id, user_id, room_id,
category_id, content):
"""Update a room entry in a group summary
"""
if category_id:
path = PREFIX + "/groups/%s/summary/categories/%s/rooms/%s" % (
group_id, category_id, room_id,
)
else:
path = PREFIX + "/groups/%s/summary/rooms/%s" % (group_id, room_id,)
return self.client.post_json(
destination=destination,
path=path,
args={"requester_user_id": user_id},
data=content,
ignore_backoff=True,
)
@log_function
def delete_group_summary_room(self, destination, group_id, user_id, room_id,
category_id):
"""Delete a room entry in a group summary
"""
if category_id:
path = PREFIX + "/groups/%s/summary/categories/%s/rooms/%s" % (
group_id, category_id, room_id,
)
else:
path = PREFIX + "/groups/%s/summary/rooms/%s" % (group_id, room_id,)
return self.client.delete_json(
destination=destination,
path=path,
args={"requester_user_id": user_id},
ignore_backoff=True,
)
@log_function
def get_group_categories(self, destination, group_id, requester_user_id):
"""Get all categories in a group
"""
path = PREFIX + "/groups/%s/categories" % (group_id,)
return self.client.get_json(
destination=destination,
path=path,
args={"requester_user_id": requester_user_id},
ignore_backoff=True,
)
@log_function
def get_group_category(self, destination, group_id, requester_user_id, category_id):
"""Get category info in a group
"""
path = PREFIX + "/groups/%s/categories/%s" % (group_id, category_id,)
return self.client.get_json(
destination=destination,
path=path,
args={"requester_user_id": requester_user_id},
ignore_backoff=True,
)
@log_function
def update_group_category(self, destination, group_id, requester_user_id, category_id,
content):
"""Update a category in a group
"""
path = PREFIX + "/groups/%s/categories/%s" % (group_id, category_id,)
return self.client.post_json(
destination=destination,
path=path,
args={"requester_user_id": requester_user_id},
data=content,
ignore_backoff=True,
)
@log_function
def delete_group_category(self, destination, group_id, requester_user_id,
category_id):
"""Delete a category in a group
"""
path = PREFIX + "/groups/%s/categories/%s" % (group_id, category_id,)
return self.client.delete_json(
destination=destination,
path=path,
args={"requester_user_id": requester_user_id},
ignore_backoff=True,
)
@log_function
def get_group_roles(self, destination, group_id, requester_user_id):
"""Get all roles in a group
"""
path = PREFIX + "/groups/%s/roles" % (group_id,)
return self.client.get_json(
destination=destination,
path=path,
args={"requester_user_id": requester_user_id},
ignore_backoff=True,
)
@log_function
def get_group_role(self, destination, group_id, requester_user_id, role_id):
"""Get a roles info
"""
path = PREFIX + "/groups/%s/roles/%s" % (group_id, role_id,)
return self.client.get_json(
destination=destination,
path=path,
args={"requester_user_id": requester_user_id},
ignore_backoff=True,
)
@log_function
def update_group_role(self, destination, group_id, requester_user_id, role_id,
content):
"""Update a role in a group
"""
path = PREFIX + "/groups/%s/roles/%s" % (group_id, role_id,)
return self.client.post_json(
destination=destination,
path=path,
args={"requester_user_id": requester_user_id},
data=content,
ignore_backoff=True,
)
@log_function
def delete_group_role(self, destination, group_id, requester_user_id, role_id):
"""Delete a role in a group
"""
path = PREFIX + "/groups/%s/roles/%s" % (group_id, role_id,)
return self.client.delete_json(
destination=destination,
path=path,
args={"requester_user_id": requester_user_id},
ignore_backoff=True,
)
@log_function
def update_group_summary_user(self, destination, group_id, requester_user_id,
user_id, role_id, content):
"""Update a users entry in a group
"""
if role_id:
path = PREFIX + "/groups/%s/summary/roles/%s/users/%s" % (
group_id, role_id, user_id,
)
else:
path = PREFIX + "/groups/%s/summary/users/%s" % (group_id, user_id,)
return self.client.post_json(
destination=destination,
path=path,
args={"requester_user_id": requester_user_id},
data=content,
ignore_backoff=True,
)
@log_function
def delete_group_summary_user(self, destination, group_id, requester_user_id,
user_id, role_id):
"""Delete a users entry in a group
"""
if role_id:
path = PREFIX + "/groups/%s/summary/roles/%s/users/%s" % (
group_id, role_id, user_id,
)
else:
path = PREFIX + "/groups/%s/summary/users/%s" % (group_id, user_id,)
return self.client.delete_json(
destination=destination,
path=path,
args={"requester_user_id": requester_user_id},
ignore_backoff=True,
)
def bulk_get_publicised_groups(self, destination, user_ids):
"""Get the groups a list of users are publicising
"""
path = PREFIX + "/get_groups_publicised"
content = {"user_ids": user_ids}
return self.client.post_json(
destination=destination,
path=path,
data=content,
ignore_backoff=True,
)

525
synapse/federation/transport/server.py
View File

@ -25,7 +25,7 @@ from synapse.http.servlet import (
from synapse.util.ratelimitutils import FederationRateLimiter
from synapse.util.versionstring import get_version_string
from synapse.util.logcontext import preserve_fn
from synapse.types import ThirdPartyInstanceID
from synapse.types import ThirdPartyInstanceID, get_domain_from_id
import functools
import logging
@ -609,6 +609,475 @@ class FederationVersionServlet(BaseFederationServlet):
}))
class FederationGroupsProfileServlet(BaseFederationServlet):
"""Get the basic profile of a group on behalf of a user
"""
PATH = "/groups/(?P<group_id>[^/]*)/profile$"
@defer.inlineCallbacks
def on_GET(self, origin, content, query, group_id):
requester_user_id = parse_string_from_args(query, "requester_user_id")
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")
new_content = yield self.handler.get_group_profile(
group_id, requester_user_id
)
defer.returnValue((200, new_content))
class FederationGroupsSummaryServlet(BaseFederationServlet):
PATH = "/groups/(?P<group_id>[^/]*)/summary$"
@defer.inlineCallbacks
def on_GET(self, origin, content, query, group_id):
requester_user_id = parse_string_from_args(query, "requester_user_id")
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")
new_content = yield self.handler.get_group_summary(
group_id, requester_user_id
)
defer.returnValue((200, new_content))
@defer.inlineCallbacks
def on_POST(self, origin, content, query, group_id):
requester_user_id = parse_string_from_args(query, "requester_user_id")
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")
new_content = yield self.handler.update_group_profile(
group_id, requester_user_id, content
)
defer.returnValue((200, new_content))
class FederationGroupsRoomsServlet(BaseFederationServlet):
"""Get the rooms in a group on behalf of a user
"""
PATH = "/groups/(?P<group_id>[^/]*)/rooms$"
@defer.inlineCallbacks
def on_GET(self, origin, content, query, group_id):
requester_user_id = parse_string_from_args(query, "requester_user_id")
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")
new_content = yield self.handler.get_rooms_in_group(
group_id, requester_user_id
)
defer.returnValue((200, new_content))
class FederationGroupsAddRoomsServlet(BaseFederationServlet):
"""Add/remove room from group
"""
PATH = "/groups/(?P<group_id>[^/]*)/room/(?<room_id>)$"
@defer.inlineCallbacks
def on_POST(self, origin, content, query, group_id, room_id):
requester_user_id = parse_string_from_args(query, "requester_user_id")
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")
new_content = yield self.handler.add_room_to_group(
group_id, requester_user_id, room_id, content
)
defer.returnValue((200, new_content))
@defer.inlineCallbacks
def on_DELETE(self, origin, content, query, group_id, room_id):
requester_user_id = parse_string_from_args(query, "requester_user_id")
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")
new_content = yield self.handler.remove_room_from_group(
group_id, requester_user_id, room_id,
)
defer.returnValue((200, new_content))
class FederationGroupsUsersServlet(BaseFederationServlet):
"""Get the users in a group on behalf of a user
"""
PATH = "/groups/(?P<group_id>[^/]*)/users$"
@defer.inlineCallbacks
def on_GET(self, origin, content, query, group_id):
requester_user_id = parse_string_from_args(query, "requester_user_id")
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")
new_content = yield self.handler.get_users_in_group(
group_id, requester_user_id
)
defer.returnValue((200, new_content))
class FederationGroupsInviteServlet(BaseFederationServlet):
"""Ask a group server to invite someone to the group
"""
PATH = "/groups/(?P<group_id>[^/]*)/users/(?P<user_id>[^/]*)/invite$"
@defer.inlineCallbacks
def on_POST(self, origin, content, query, group_id, user_id):
requester_user_id = parse_string_from_args(query, "requester_user_id")
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")
new_content = yield self.handler.invite_to_group(
group_id, user_id, requester_user_id, content,
)
defer.returnValue((200, new_content))
class FederationGroupsAcceptInviteServlet(BaseFederationServlet):
"""Accept an invitation from the group server
"""
PATH = "/groups/(?P<group_id>[^/]*)/users/(?P<user_id>[^/]*)/accept_invite$"
@defer.inlineCallbacks
def on_POST(self, origin, content, query, group_id, user_id):
if get_domain_from_id(user_id) != origin:
raise SynapseError(403, "user_id doesn't match origin")
new_content = yield self.handler.accept_invite(
group_id, user_id, content,
)
defer.returnValue((200, new_content))
class FederationGroupsRemoveUserServlet(BaseFederationServlet):
"""Leave or kick a user from the group
"""
PATH = "/groups/(?P<group_id>[^/]*)/users/(?P<user_id>[^/]*)/remove$"
@defer.inlineCallbacks
def on_POST(self, origin, content, query, group_id, user_id):
requester_user_id = parse_string_from_args(query, "requester_user_id")
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")
new_content = yield self.handler.remove_user_from_group(
group_id, user_id, requester_user_id, content,
)
defer.returnValue((200, new_content))
class FederationGroupsLocalInviteServlet(BaseFederationServlet):
"""A group server has invited a local user
"""
PATH = "/groups/local/(?P<group_id>[^/]*)/users/(?P<user_id>[^/]*)/invite$"
@defer.inlineCallbacks
def on_POST(self, origin, content, query, group_id, user_id):
if get_domain_from_id(group_id) != origin:
raise SynapseError(403, "group_id doesn't match origin")
new_content = yield self.handler.on_invite(
group_id, user_id, content,
)
defer.returnValue((200, new_content))
class FederationGroupsRemoveLocalUserServlet(BaseFederationServlet):
"""A group server has removed a local user
"""
PATH = "/groups/local/(?P<group_id>[^/]*)/users/(?P<user_id>[^/]*)/remove$"
@defer.inlineCallbacks
def on_POST(self, origin, content, query, group_id, user_id):
if get_domain_from_id(group_id) != origin:
raise SynapseError(403, "user_id doesn't match origin")
new_content = yield self.handler.user_removed_from_group(
group_id, user_id, content,
)
defer.returnValue((200, new_content))
class FederationGroupsRenewAttestaionServlet(BaseFederationServlet):
"""A group or user's server renews their attestation
"""
PATH = "/groups/(?P<group_id>[^/]*)/renew_attestation/(?P<user_id>[^/]*)$"
@defer.inlineCallbacks
def on_POST(self, origin, content, query, group_id, user_id):
# We don't need to check auth here as we check the attestation signatures
new_content = yield self.handler.on_renew_group_attestation(
origin, content, group_id, user_id
)
defer.returnValue((200, new_content))
class FederationGroupsSummaryRoomsServlet(BaseFederationServlet):
"""Add/remove a room from the group summary, with optional category.
Matches both:
- /groups/:group/summary/rooms/:room_id
- /groups/:group/summary/categories/:category/rooms/:room_id
"""
PATH = (
"/groups/(?P<group_id>[^/]*)/summary"
"(/categories/(?P<category_id>[^/]+))?"
"/rooms/(?P<room_id>[^/]*)$"
)
@defer.inlineCallbacks
def on_POST(self, origin, content, query, group_id, category_id, room_id):
requester_user_id = parse_string_from_args(query, "requester_user_id")
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")
if category_id == "":
raise SynapseError(400, "category_id cannot be empty string")
resp = yield self.handler.update_group_summary_room(
group_id, requester_user_id,
room_id=room_id,
category_id=category_id,
content=content,
)
defer.returnValue((200, resp))
@defer.inlineCallbacks
def on_DELETE(self, origin, content, query, group_id, category_id, room_id):
requester_user_id = parse_string_from_args(query, "requester_user_id")
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")
if category_id == "":
raise SynapseError(400, "category_id cannot be empty string")
resp = yield self.handler.delete_group_summary_room(
group_id, requester_user_id,
room_id=room_id,
category_id=category_id,
)
defer.returnValue((200, resp))
class FederationGroupsCategoriesServlet(BaseFederationServlet):
"""Get all categories for a group
"""
PATH = (
"/groups/(?P<group_id>[^/]*)/categories/$"
)
@defer.inlineCallbacks
def on_GET(self, origin, content, query, group_id):
requester_user_id = parse_string_from_args(query, "requester_user_id")
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")
resp = yield self.handler.get_group_categories(
group_id, requester_user_id,
)
defer.returnValue((200, resp))
class FederationGroupsCategoryServlet(BaseFederationServlet):
"""Add/remove/get a category in a group
"""
PATH = (
"/groups/(?P<group_id>[^/]*)/categories/(?P<category_id>[^/]+)$"
)
@defer.inlineCallbacks
def on_GET(self, origin, content, query, group_id, category_id):
requester_user_id = parse_string_from_args(query, "requester_user_id")
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")
resp = yield self.handler.get_group_category(
group_id, requester_user_id, category_id
)
defer.returnValue((200, resp))
@defer.inlineCallbacks
def on_POST(self, origin, content, query, group_id, category_id):
requester_user_id = parse_string_from_args(query, "requester_user_id")
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")
if category_id == "":
raise SynapseError(400, "category_id cannot be empty string")
resp = yield self.handler.upsert_group_category(
group_id, requester_user_id, category_id, content,
)
defer.returnValue((200, resp))
@defer.inlineCallbacks
def on_DELETE(self, origin, content, query, group_id, category_id):
requester_user_id = parse_string_from_args(query, "requester_user_id")
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")
if category_id == "":
raise SynapseError(400, "category_id cannot be empty string")
resp = yield self.handler.delete_group_category(
group_id, requester_user_id, category_id,
)
defer.returnValue((200, resp))
class FederationGroupsRolesServlet(BaseFederationServlet):
"""Get roles in a group
"""
PATH = (
"/groups/(?P<group_id>[^/]*)/roles/$"
)
@defer.inlineCallbacks
def on_GET(self, origin, content, query, group_id):
requester_user_id = parse_string_from_args(query, "requester_user_id")
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")
resp = yield self.handler.get_group_roles(
group_id, requester_user_id,
)
defer.returnValue((200, resp))
class FederationGroupsRoleServlet(BaseFederationServlet):
"""Add/remove/get a role in a group
"""
PATH = (
"/groups/(?P<group_id>[^/]*)/roles/(?P<role_id>[^/]+)$"
)
@defer.inlineCallbacks
def on_GET(self, origin, content, query, group_id, role_id):
requester_user_id = parse_string_from_args(query, "requester_user_id")
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")
resp = yield self.handler.get_group_role(
group_id, requester_user_id, role_id
)
defer.returnValue((200, resp))
@defer.inlineCallbacks
def on_POST(self, origin, content, query, group_id, role_id):
requester_user_id = parse_string_from_args(query, "requester_user_id")
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")
if role_id == "":
raise SynapseError(400, "role_id cannot be empty string")
resp = yield self.handler.update_group_role(
group_id, requester_user_id, role_id, content,
)
defer.returnValue((200, resp))
@defer.inlineCallbacks
def on_DELETE(self, origin, content, query, group_id, role_id):
requester_user_id = parse_string_from_args(query, "requester_user_id")
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")
if role_id == "":
raise SynapseError(400, "role_id cannot be empty string")
resp = yield self.handler.delete_group_role(
group_id, requester_user_id, role_id,
)
defer.returnValue((200, resp))
class FederationGroupsSummaryUsersServlet(BaseFederationServlet):
"""Add/remove a user from the group summary, with optional role.
Matches both:
- /groups/:group/summary/users/:user_id
- /groups/:group/summary/roles/:role/users/:user_id
"""
PATH = (
"/groups/(?P<group_id>[^/]*)/summary"
"(/roles/(?P<role_id>[^/]+))?"
"/users/(?P<user_id>[^/]*)$"
)
@defer.inlineCallbacks
def on_POST(self, origin, content, query, group_id, role_id, user_id):
requester_user_id = parse_string_from_args(query, "requester_user_id")
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")
if role_id == "":
raise SynapseError(400, "role_id cannot be empty string")
resp = yield self.handler.update_group_summary_user(
group_id, requester_user_id,
user_id=user_id,
role_id=role_id,
content=content,
)
defer.returnValue((200, resp))
@defer.inlineCallbacks
def on_DELETE(self, origin, content, query, group_id, role_id, user_id):
requester_user_id = parse_string_from_args(query, "requester_user_id")
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")
if role_id == "":
raise SynapseError(400, "role_id cannot be empty string")
resp = yield self.handler.delete_group_summary_user(
group_id, requester_user_id,
user_id=user_id,
role_id=role_id,
)
defer.returnValue((200, resp))
class FederationGroupsBulkPublicisedServlet(BaseFederationServlet):
"""Get roles in a group
"""
PATH = (
"/get_groups_publicised$"
)
@defer.inlineCallbacks
def on_POST(self, origin, content, query):
resp = yield self.handler.bulk_get_publicised_groups(
content["user_ids"], proxy=False,
)
defer.returnValue((200, resp))
FEDERATION_SERVLET_CLASSES = (
FederationSendServlet,
FederationPullServlet,
@ -635,11 +1104,41 @@ FEDERATION_SERVLET_CLASSES = (
FederationVersionServlet,
)
ROOM_LIST_CLASSES = (
PublicRoomList,
)
GROUP_SERVER_SERVLET_CLASSES = (
FederationGroupsProfileServlet,
FederationGroupsSummaryServlet,
FederationGroupsRoomsServlet,
FederationGroupsUsersServlet,
FederationGroupsInviteServlet,
FederationGroupsAcceptInviteServlet,
FederationGroupsRemoveUserServlet,
FederationGroupsSummaryRoomsServlet,
FederationGroupsCategoriesServlet,
FederationGroupsCategoryServlet,
FederationGroupsRolesServlet,
FederationGroupsRoleServlet,
FederationGroupsSummaryUsersServlet,
)
GROUP_LOCAL_SERVLET_CLASSES = (
FederationGroupsLocalInviteServlet,
FederationGroupsRemoveLocalUserServlet,
FederationGroupsBulkPublicisedServlet,
)
GROUP_ATTESTATION_SERVLET_CLASSES = (
FederationGroupsRenewAttestaionServlet,
)
def register_servlets(hs, resource, authenticator, ratelimiter):
for servletclass in FEDERATION_SERVLET_CLASSES:
servletclass(
@ -656,3 +1155,27 @@ def register_servlets(hs, resource, authenticator, ratelimiter):
ratelimiter=ratelimiter,
server_name=hs.hostname,
).register(resource)
for servletclass in GROUP_SERVER_SERVLET_CLASSES:
servletclass(
handler=hs.get_groups_server_handler(),
authenticator=authenticator,
ratelimiter=ratelimiter,
server_name=hs.hostname,
).register(resource)
for servletclass in GROUP_LOCAL_SERVLET_CLASSES:
servletclass(
handler=hs.get_groups_local_handler(),
authenticator=authenticator,
ratelimiter=ratelimiter,
server_name=hs.hostname,
).register(resource)
for servletclass in GROUP_ATTESTATION_SERVLET_CLASSES:
servletclass(
handler=hs.get_groups_attestation_renewer(),
authenticator=authenticator,
ratelimiter=ratelimiter,
server_name=hs.hostname,
).register(resource)

0
synapse/groups/__init__.py Normal file
View File

150
synapse/groups/attestations.py Normal file
View File

@ -0,0 +1,150 @@
# -*- coding: utf-8 -*-
# Copyright 2017 Vector Creations 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 SynapseError
from synapse.types import get_domain_from_id
from synapse.util.logcontext import preserve_fn
from signedjson.sign import sign_json
# Default validity duration for new attestations we create
DEFAULT_ATTESTATION_LENGTH_MS = 3 * 24 * 60 * 60 * 1000
# Start trying to update our attestations when they come this close to expiring
UPDATE_ATTESTATION_TIME_MS = 1 * 24 * 60 * 60 * 1000
class GroupAttestationSigning(object):
"""Creates and verifies group attestations.
"""
def __init__(self, hs):
self.keyring = hs.get_keyring()
self.clock = hs.get_clock()
self.server_name = hs.hostname
self.signing_key = hs.config.signing_key[0]
@defer.inlineCallbacks
def verify_attestation(self, attestation, group_id, user_id, server_name=None):
"""Verifies that the given attestation matches the given parameters.
An optional server_name can be supplied to explicitly set which server's
signature is expected. Otherwise assumes that either the group_id or user_id
is local and uses the other's server as the one to check.
"""
if not server_name:
if get_domain_from_id(group_id) == self.server_name:
server_name = get_domain_from_id(user_id)
elif get_domain_from_id(user_id) == self.server_name:
server_name = get_domain_from_id(group_id)
else:
raise Exception("Expected either group_id or user_id to be local")
if user_id != attestation["user_id"]:
raise SynapseError(400, "Attestation has incorrect user_id")
if group_id != attestation["group_id"]:
raise SynapseError(400, "Attestation has incorrect group_id")
valid_until_ms = attestation["valid_until_ms"]
# TODO: We also want to check that *new* attestations that people give
# us to store are valid for at least a little while.
if valid_until_ms < self.clock.time_msec():
raise SynapseError(400, "Attestation expired")
yield self.keyring.verify_json_for_server(server_name, attestation)
def create_attestation(self, group_id, user_id):
"""Create an attestation for the group_id and user_id with default
validity length.
"""
return sign_json({
"group_id": group_id,
"user_id": user_id,
"valid_until_ms": self.clock.time_msec() + DEFAULT_ATTESTATION_LENGTH_MS,
}, self.server_name, self.signing_key)
class GroupAttestionRenewer(object):
"""Responsible for sending and receiving attestation updates.
"""
def __init__(self, hs):
self.clock = hs.get_clock()
self.store = hs.get_datastore()
self.assestations = hs.get_groups_attestation_signing()
self.transport_client = hs.get_federation_transport_client()
self.is_mine_id = hs.is_mine_id
self._renew_attestations_loop = self.clock.looping_call(
self._renew_attestations, 30 * 60 * 1000,
)
@defer.inlineCallbacks
def on_renew_attestation(self, group_id, user_id, content):
"""When a remote updates an attestation
"""
attestation = content["attestation"]
if not self.is_mine_id(group_id) and not self.is_mine_id(user_id):
raise SynapseError(400, "Neither user not group are on this server")
yield self.attestations.verify_attestation(
attestation,
user_id=user_id,
group_id=group_id,
)
yield self.store.update_remote_attestion(group_id, user_id, attestation)
defer.returnValue({})
@defer.inlineCallbacks
def _renew_attestations(self):
"""Called periodically to check if we need to update any of our attestations
"""
now = self.clock.time_msec()
rows = yield self.store.get_attestations_need_renewals(
now + UPDATE_ATTESTATION_TIME_MS
)
@defer.inlineCallbacks
def _renew_attestation(self, group_id, user_id):
attestation = self.attestations.create_attestation(group_id, user_id)
if self.hs.is_mine_id(group_id):
destination = get_domain_from_id(user_id)
else:
destination = get_domain_from_id(group_id)
yield self.transport_client.renew_group_attestation(
destination, group_id, user_id,
content={"attestation": attestation},
)
yield self.store.update_attestation_renewal(
group_id, user_id, attestation
)
for row in rows:
group_id = row["group_id"]
user_id = row["user_id"]
preserve_fn(_renew_attestation)(group_id, user_id)

741
synapse/groups/groups_server.py Normal file
View File

@ -0,0 +1,741 @@
# -*- coding: utf-8 -*-
# Copyright 2017 Vector Creations 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 SynapseError
from synapse.types import UserID, get_domain_from_id, RoomID
import logging
logger = logging.getLogger(__name__)
# TODO: Allow users to "knock" or simpkly join depending on rules
# TODO: Federation admin APIs
# TODO: is_priveged flag to users and is_public to users and rooms
# TODO: Audit log for admins (profile updates, membership changes, users who tried
# to join but were rejected, etc)
# TODO: Flairs
class GroupsServerHandler(object):
def __init__(self, hs):
self.hs = hs
self.store = hs.get_datastore()
self.room_list_handler = hs.get_room_list_handler()
self.auth = hs.get_auth()
self.clock = hs.get_clock()
self.keyring = hs.get_keyring()
self.is_mine_id = hs.is_mine_id
self.signing_key = hs.config.signing_key[0]
self.server_name = hs.hostname
self.attestations = hs.get_groups_attestation_signing()
self.transport_client = hs.get_federation_transport_client()
self.profile_handler = hs.get_profile_handler()
# Ensure attestations get renewed
hs.get_groups_attestation_renewer()
@defer.inlineCallbacks
def check_group_is_ours(self, group_id, and_exists=False, and_is_admin=None):
"""Check that the group is ours, and optionally if it exists.
If group does exist then return group.
Args:
group_id (str)
and_exists (bool): whether to also check if group exists
and_is_admin (str): whether to also check if given str is a user_id
that is an admin
"""
if not self.is_mine_id(group_id):
raise SynapseError(400, "Group not on this server")
group = yield self.store.get_group(group_id)
if and_exists and not group:
raise SynapseError(404, "Unknown group")
if and_is_admin:
is_admin = yield self.store.is_user_admin_in_group(group_id, and_is_admin)
if not is_admin:
raise SynapseError(403, "User is not admin in group")
defer.returnValue(group)
@defer.inlineCallbacks
def get_group_summary(self, group_id, requester_user_id):
"""Get the summary for a group as seen by requester_user_id.
The group summary consists of the profile of the room, and a curated
list of users and rooms. These list *may* be organised by role/category.
The roles/categories are ordered, and so are the users/rooms within them.
A user/room may appear in multiple roles/categories.
"""
yield self.check_group_is_ours(group_id, and_exists=True)
is_user_in_group = yield self.store.is_user_in_group(requester_user_id, group_id)
profile = yield self.get_group_profile(group_id, requester_user_id)
users, roles = yield self.store.get_users_for_summary_by_role(
group_id, include_private=is_user_in_group,
)
# TODO: Add profiles to users
rooms, categories = yield self.store.get_rooms_for_summary_by_category(
group_id, include_private=is_user_in_group,
)
for room_entry in rooms:
room_id = room_entry["room_id"]
joined_users = yield self.store.get_users_in_room(room_id)
entry = yield self.room_list_handler.generate_room_entry(
room_id, len(joined_users),
with_alias=False, allow_private=True,
)
entry = dict(entry) # so we don't change whats cached
entry.pop("room_id", None)
room_entry["profile"] = entry
rooms.sort(key=lambda e: e.get("order", 0))
for entry in users:
user_id = entry["user_id"]
if not self.is_mine_id(requester_user_id):
attestation = yield self.store.get_remote_attestation(group_id, user_id)
if not attestation:
continue
entry["attestation"] = attestation
else:
entry["attestation"] = self.attestations.create_attestation(
group_id, user_id,
)
user_profile = yield self.profile_handler.get_profile_from_cache(user_id)
entry.update(user_profile)
users.sort(key=lambda e: e.get("order", 0))
membership_info = yield self.store.get_users_membership_info_in_group(
group_id, requester_user_id,
)
defer.returnValue({
"profile": profile,
"users_section": {
"users": users,
"roles": roles,
"total_user_count_estimate": 0, # TODO
},
"rooms_section": {
"rooms": rooms,
"categories": categories,
"total_room_count_estimate": 0, # TODO
},
"user": membership_info,
})
@defer.inlineCallbacks
def update_group_summary_room(self, group_id, user_id, room_id, category_id, content):
"""Add/update a room to the group summary
"""
yield self.check_group_is_ours(group_id, and_exists=True, and_is_admin=user_id)
RoomID.from_string(room_id) # Ensure valid room id
order = content.get("order", None)
is_public = _parse_visibility_from_contents(content)
yield self.store.add_room_to_summary(
group_id=group_id,
room_id=room_id,
category_id=category_id,
order=order,
is_public=is_public,
)
defer.returnValue({})
@defer.inlineCallbacks
def delete_group_summary_room(self, group_id, user_id, room_id, category_id):
"""Remove a room from the summary
"""
yield self.check_group_is_ours(group_id, and_exists=True, and_is_admin=user_id)
yield self.store.remove_room_from_summary(
group_id=group_id,
room_id=room_id,
category_id=category_id,
)
defer.returnValue({})
@defer.inlineCallbacks
def get_group_categories(self, group_id, user_id):
"""Get all categories in a group (as seen by user)
"""
yield self.check_group_is_ours(group_id, and_exists=True)
categories = yield self.store.get_group_categories(
group_id=group_id,
)
defer.returnValue({"categories": categories})
@defer.inlineCallbacks
def get_group_category(self, group_id, user_id, category_id):
"""Get a specific category in a group (as seen by user)
"""
yield self.check_group_is_ours(group_id, and_exists=True)
res = yield self.store.get_group_category(
group_id=group_id,
category_id=category_id,
)
defer.returnValue(res)
@defer.inlineCallbacks
def update_group_category(self, group_id, user_id, category_id, content):
"""Add/Update a group category
"""
yield self.check_group_is_ours(group_id, and_exists=True, and_is_admin=user_id)
is_public = _parse_visibility_from_contents(content)
profile = content.get("profile")
yield self.store.upsert_group_category(
group_id=group_id,
category_id=category_id,
is_public=is_public,
profile=profile,
)
defer.returnValue({})
@defer.inlineCallbacks
def delete_group_category(self, group_id, user_id, category_id):
"""Delete a group category
"""
yield self.check_group_is_ours(group_id, and_exists=True, and_is_admin=user_id)
yield self.store.remove_group_category(
group_id=group_id,
category_id=category_id,
)
defer.returnValue({})
@defer.inlineCallbacks
def get_group_roles(self, group_id, user_id):
"""Get all roles in a group (as seen by user)
"""
yield self.check_group_is_ours(group_id, and_exists=True)
roles = yield self.store.get_group_roles(
group_id=group_id,
)
defer.returnValue({"roles": roles})
@defer.inlineCallbacks
def get_group_role(self, group_id, user_id, role_id):
"""Get a specific role in a group (as seen by user)
"""
yield self.check_group_is_ours(group_id, and_exists=True)
res = yield self.store.get_group_role(
group_id=group_id,
role_id=role_id,
)
defer.returnValue(res)
@defer.inlineCallbacks
def update_group_role(self, group_id, user_id, role_id, content):
"""Add/update a role in a group
"""
yield self.check_group_is_ours(group_id, and_exists=True, and_is_admin=user_id)
is_public = _parse_visibility_from_contents(content)
profile = content.get("profile")
yield self.store.upsert_group_role(
group_id=group_id,
role_id=role_id,
is_public=is_public,
profile=profile,
)
defer.returnValue({})
@defer.inlineCallbacks
def delete_group_role(self, group_id, user_id, role_id):
"""Remove role from group
"""
yield self.check_group_is_ours(group_id, and_exists=True, and_is_admin=user_id)
yield self.store.remove_group_role(
group_id=group_id,
role_id=role_id,
)
defer.returnValue({})
@defer.inlineCallbacks
def update_group_summary_user(self, group_id, requester_user_id, user_id, role_id,
content):
"""Add/update a users entry in the group summary
"""
yield self.check_group_is_ours(
group_id, and_exists=True, and_is_admin=requester_user_id,
)
order = content.get("order", None)
is_public = _parse_visibility_from_contents(content)
yield self.store.add_user_to_summary(
group_id=group_id,
user_id=user_id,
role_id=role_id,
order=order,
is_public=is_public,
)
defer.returnValue({})
@defer.inlineCallbacks
def delete_group_summary_user(self, group_id, requester_user_id, user_id, role_id):
"""Remove a user from the group summary
"""
yield self.check_group_is_ours(
group_id, and_exists=True, and_is_admin=requester_user_id,
)
yield self.store.remove_user_from_summary(
group_id=group_id,
user_id=user_id,
role_id=role_id,
)
defer.returnValue({})
@defer.inlineCallbacks
def get_group_profile(self, group_id, requester_user_id):
"""Get the group profile as seen by requester_user_id
"""
yield self.check_group_is_ours(group_id)
group_description = yield self.store.get_group(group_id)
if group_description:
defer.returnValue(group_description)
else:
raise SynapseError(404, "Unknown group")
@defer.inlineCallbacks
def update_group_profile(self, group_id, requester_user_id, content):
"""Update the group profile
"""
yield self.check_group_is_ours(
group_id, and_exists=True, and_is_admin=requester_user_id,
)
profile = {}
for keyname in ("name", "avatar_url", "short_description",
"long_description"):
if keyname in content:
value = content[keyname]
if not isinstance(value, basestring):
raise SynapseError(400, "%r value is not a string" % (keyname,))
profile[keyname] = value
yield self.store.update_group_profile(group_id, profile)
@defer.inlineCallbacks
def get_users_in_group(self, group_id, requester_user_id):
"""Get the users in group as seen by requester_user_id.
The ordering is arbitrary at the moment
"""
yield self.check_group_is_ours(group_id, and_exists=True)
is_user_in_group = yield self.store.is_user_in_group(requester_user_id, group_id)
user_results = yield self.store.get_users_in_group(
group_id, include_private=is_user_in_group,
)
chunk = []
for user_result in user_results:
g_user_id = user_result["user_id"]
is_public = user_result["is_public"]
entry = {"user_id": g_user_id}
profile = yield self.profile_handler.get_profile_from_cache(g_user_id)
entry.update(profile)
if not is_public:
entry["is_public"] = False
if not self.is_mine_id(requester_user_id):
attestation = yield self.store.get_remote_attestation(group_id, g_user_id)
if not attestation:
continue
entry["attestation"] = attestation
else:
entry["attestation"] = self.attestations.create_attestation(
group_id, g_user_id,
)
chunk.append(entry)
# TODO: If admin add lists of users whose attestations have timed out
defer.returnValue({
"chunk": chunk,
"total_user_count_estimate": len(user_results),
})
@defer.inlineCallbacks
def get_rooms_in_group(self, group_id, requester_user_id):
"""Get the rooms in group as seen by requester_user_id
This returns rooms in order of decreasing number of joined users
"""
yield self.check_group_is_ours(group_id, and_exists=True)
is_user_in_group = yield self.store.is_user_in_group(requester_user_id, group_id)
room_results = yield self.store.get_rooms_in_group(
group_id, include_private=is_user_in_group,
)
chunk = []
for room_result in room_results:
room_id = room_result["room_id"]
is_public = room_result["is_public"]
joined_users = yield self.store.get_users_in_room(room_id)
entry = yield self.room_list_handler.generate_room_entry(
room_id, len(joined_users),
with_alias=False, allow_private=True,
)
if not entry:
continue
if not is_public:
entry["is_public"] = False
chunk.append(entry)
chunk.sort(key=lambda e: -e["num_joined_members"])
defer.returnValue({
"chunk": chunk,
"total_room_count_estimate": len(room_results),
})
@defer.inlineCallbacks
def add_room_to_group(self, group_id, requester_user_id, room_id, content):
"""Add room to group
"""
RoomID.from_string(room_id) # Ensure valid room id
yield self.check_group_is_ours(
group_id, and_exists=True, and_is_admin=requester_user_id
)
is_public = _parse_visibility_from_contents(content)
yield self.store.add_room_to_group(group_id, room_id, is_public=is_public)
defer.returnValue({})
@defer.inlineCallbacks
def remove_room_from_group(self, group_id, requester_user_id, room_id):
"""Remove room from group
"""
yield self.check_group_is_ours(
group_id, and_exists=True, and_is_admin=requester_user_id
)
yield self.store.remove_room_from_group(group_id, room_id)
defer.returnValue({})
@defer.inlineCallbacks
def invite_to_group(self, group_id, user_id, requester_user_id, content):
"""Invite user to group
"""
group = yield self.check_group_is_ours(
group_id, and_exists=True, and_is_admin=requester_user_id
)
# TODO: Check if user knocked
# TODO: Check if user is already invited
content = {
"profile": {
"name": group["name"],
"avatar_url": group["avatar_url"],
},
"inviter": requester_user_id,
}
if self.hs.is_mine_id(user_id):
groups_local = self.hs.get_groups_local_handler()
res = yield groups_local.on_invite(group_id, user_id, content)
local_attestation = None
else:
local_attestation = self.attestations.create_attestation(group_id, user_id)
content.update({
"attestation": local_attestation,
})
res = yield self.transport_client.invite_to_group_notification(
get_domain_from_id(user_id), group_id, user_id, content
)
user_profile = res.get("user_profile", {})
yield self.store.add_remote_profile_cache(
user_id,
displayname=user_profile.get("displayname"),
avatar_url=user_profile.get("avatar_url"),
)
if res["state"] == "join":
if not self.hs.is_mine_id(user_id):
remote_attestation = res["attestation"]
yield self.attestations.verify_attestation(
remote_attestation,
user_id=user_id,
group_id=group_id,
)
else:
remote_attestation = None
yield self.store.add_user_to_group(
group_id, user_id,
is_admin=False,
is_public=False, # TODO
local_attestation=local_attestation,
remote_attestation=remote_attestation,
)
elif res["state"] == "invite":
yield self.store.add_group_invite(
group_id, user_id,
)
defer.returnValue({
"state": "invite"
})
elif res["state"] == "reject":
defer.returnValue({
"state": "reject"
})
else:
raise SynapseError(502, "Unknown state returned by HS")
@defer.inlineCallbacks
def accept_invite(self, group_id, user_id, content):
"""User tries to accept an invite to the group.
This is different from them asking to join, and so should error if no
invite exists (and they're not a member of the group)
"""
yield self.check_group_is_ours(group_id, and_exists=True)
if not self.store.is_user_invited_to_local_group(group_id, user_id):
raise SynapseError(403, "User not invited to group")
if not self.hs.is_mine_id(user_id):
remote_attestation = content["attestation"]
yield self.attestations.verify_attestation(
remote_attestation,
user_id=user_id,
group_id=group_id,
)
else:
remote_attestation = None
local_attestation = self.attestations.create_attestation(group_id, user_id)
is_public = _parse_visibility_from_contents(content)
yield self.store.add_user_to_group(
group_id, user_id,
is_admin=False,
is_public=is_public,
local_attestation=local_attestation,
remote_attestation=remote_attestation,
)
defer.returnValue({
"state": "join",
"attestation": local_attestation,
})
@defer.inlineCallbacks
def knock(self, group_id, user_id, content):
"""A user requests becoming a member of the group
"""
yield self.check_group_is_ours(group_id, and_exists=True)
raise NotImplementedError()
@defer.inlineCallbacks
def accept_knock(self, group_id, user_id, content):
"""Accept a users knock to the room.
Errors if the user hasn't knocked, rather than inviting them.
"""
yield self.check_group_is_ours(group_id, and_exists=True)
raise NotImplementedError()
@defer.inlineCallbacks
def remove_user_from_group(self, group_id, user_id, requester_user_id, content):
"""Remove a user from the group; either a user is leaving or and admin
kicked htem.
"""
yield self.check_group_is_ours(group_id, and_exists=True)
is_kick = False
if requester_user_id != user_id:
is_admin = yield self.store.is_user_admin_in_group(
group_id, requester_user_id
)
if not is_admin:
raise SynapseError(403, "User is not admin in group")
is_kick = True
yield self.store.remove_user_from_group(
group_id, user_id,
)
if is_kick:
if self.hs.is_mine_id(user_id):
groups_local = self.hs.get_groups_local_handler()
yield groups_local.user_removed_from_group(group_id, user_id, {})
else:
yield self.transport_client.remove_user_from_group_notification(
get_domain_from_id(user_id), group_id, user_id, {}
)
if not self.hs.is_mine_id(user_id):
yield self.store.maybe_delete_remote_profile_cache(user_id)
defer.returnValue({})
@defer.inlineCallbacks
def create_group(self, group_id, user_id, content):
group = yield self.check_group_is_ours(group_id)
logger.info("Attempting to create group with ID: %r", group_id)
if group:
raise SynapseError(400, "Group already exists")
# TODO: Add config to enforce that only server admins can create rooms
is_admin = yield self.auth.is_server_admin(UserID.from_string(user_id))
if not is_admin:
raise SynapseError(403, "Only server admin can create group on this server")
profile = content.get("profile", {})
name = profile.get("name")
avatar_url = profile.get("avatar_url")
short_description = profile.get("short_description")
long_description = profile.get("long_description")
user_profile = content.get("user_profile", {})
yield self.store.create_group(
group_id,
user_id,
name=name,
avatar_url=avatar_url,
short_description=short_description,
long_description=long_description,
)
if not self.hs.is_mine_id(user_id):
remote_attestation = content["attestation"]
yield self.attestations.verify_attestation(
remote_attestation,
user_id=user_id,
group_id=group_id,
)
local_attestation = self.attestations.create_attestation(group_id, user_id)
else:
local_attestation = None
remote_attestation = None
yield self.store.add_user_to_group(
group_id, user_id,
is_admin=True,
is_public=True, # TODO
local_attestation=local_attestation,
remote_attestation=remote_attestation,
)
if not self.hs.is_mine_id(user_id):
yield self.store.add_remote_profile_cache(
user_id,
displayname=user_profile.get("displayname"),
avatar_url=user_profile.get("avatar_url"),
)
defer.returnValue({
"group_id": group_id,
})
def _parse_visibility_from_contents(content):
"""Given a content for a request parse out whether the entity should be
public or not
"""
visibility = content.get("visibility")
if visibility:
vis_type = visibility["type"]
if vis_type not in ("public", "private"):
raise SynapseError(
400, "Synapse only supports 'public'/'private' visibility"
)
is_public = vis_type == "public"
else:
is_public = True
return is_public

2
synapse/handlers/__init__.py
View File

@ -20,7 +20,6 @@ from .room import (
from .room_member import RoomMemberHandler
from .message import MessageHandler
from .federation import FederationHandler
from .profile import ProfileHandler
from .directory import DirectoryHandler
from .admin import AdminHandler
from .identity import IdentityHandler
@ -52,7 +51,6 @@ class Handlers(object):
self.room_creation_handler = RoomCreationHandler(hs)
self.room_member_handler = RoomMemberHandler(hs)
self.federation_handler = FederationHandler(hs)
self.profile_handler = ProfileHandler(hs)
self.directory_handler = DirectoryHandler(hs)
self.admin_handler = AdminHandler(hs)
self.identity_handler = IdentityHandler(hs)

402
synapse/handlers/groups_local.py Normal file
View File

@ -0,0 +1,402 @@
# -*- coding: utf-8 -*-
# Copyright 2017 Vector Creations 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 SynapseError
from synapse.types import get_domain_from_id
import logging
logger = logging.getLogger(__name__)
def _create_rerouter(func_name):
"""Returns a function that looks at the group id and calls the function
on federation or the local group server if the group is local
"""
def f(self, group_id, *args, **kwargs):
if self.is_mine_id(group_id):
return getattr(self.groups_server_handler, func_name)(
group_id, *args, **kwargs
)
else:
destination = get_domain_from_id(group_id)
return getattr(self.transport_client, func_name)(
destination, group_id, *args, **kwargs
)
return f
class GroupsLocalHandler(object):
def __init__(self, hs):
self.hs = hs
self.store = hs.get_datastore()
self.room_list_handler = hs.get_room_list_handler()
self.groups_server_handler = hs.get_groups_server_handler()
self.transport_client = hs.get_federation_transport_client()
self.auth = hs.get_auth()
self.clock = hs.get_clock()
self.keyring = hs.get_keyring()
self.is_mine_id = hs.is_mine_id
self.signing_key = hs.config.signing_key[0]
self.server_name = hs.hostname
self.notifier = hs.get_notifier()
self.attestations = hs.get_groups_attestation_signing()
self.profile_handler = hs.get_profile_handler()
# Ensure attestations get renewed
hs.get_groups_attestation_renewer()
# The following functions merely route the query to the local groups server
# or federation depending on if the group is local or remote
get_group_profile = _create_rerouter("get_group_profile")
update_group_profile = _create_rerouter("update_group_profile")
get_rooms_in_group = _create_rerouter("get_rooms_in_group")
add_room_to_group = _create_rerouter("add_room_to_group")
remove_room_from_group = _create_rerouter("remove_room_from_group")
update_group_summary_room = _create_rerouter("update_group_summary_room")
delete_group_summary_room = _create_rerouter("delete_group_summary_room")
update_group_category = _create_rerouter("update_group_category")
delete_group_category = _create_rerouter("delete_group_category")
get_group_category = _create_rerouter("get_group_category")
get_group_categories = _create_rerouter("get_group_categories")
update_group_summary_user = _create_rerouter("update_group_summary_user")
delete_group_summary_user = _create_rerouter("delete_group_summary_user")
update_group_role = _create_rerouter("update_group_role")
delete_group_role = _create_rerouter("delete_group_role")
get_group_role = _create_rerouter("get_group_role")
get_group_roles = _create_rerouter("get_group_roles")
@defer.inlineCallbacks
def get_group_summary(self, group_id, requester_user_id):
"""Get the group summary for a group.
If the group is remote we check that the users have valid attestations.
"""
if self.is_mine_id(group_id):
res = yield self.groups_server_handler.get_group_summary(
group_id, requester_user_id
)
else:
res = yield self.transport_client.get_group_summary(
get_domain_from_id(group_id), group_id, requester_user_id,
)
# Loop through the users and validate the attestations.
chunk = res["users_section"]["users"]
valid_users = []
for entry in chunk:
g_user_id = entry["user_id"]
attestation = entry.pop("attestation")
try:
yield self.attestations.verify_attestation(
attestation,
group_id=group_id,
user_id=g_user_id,
)
valid_users.append(entry)
except Exception as e:
logger.info("Failed to verify user is in group: %s", e)
res["users_section"]["users"] = valid_users
res["users_section"]["users"].sort(key=lambda e: e.get("order", 0))
res["rooms_section"]["rooms"].sort(key=lambda e: e.get("order", 0))
# Add `is_publicised` flag to indicate whether the user has publicised their
# membership of the group on their profile
result = yield self.store.get_publicised_groups_for_user(requester_user_id)
is_publicised = group_id in result
res.setdefault("user", {})["is_publicised"] = is_publicised
defer.returnValue(res)
@defer.inlineCallbacks
def create_group(self, group_id, user_id, content):
"""Create a group
"""
logger.info("Asking to create group with ID: %r", group_id)
if self.is_mine_id(group_id):
res = yield self.groups_server_handler.create_group(
group_id, user_id, content
)
local_attestation = None
remote_attestation = None
else:
local_attestation = self.attestations.create_attestation(group_id, user_id)
content["attestation"] = local_attestation
content["user_profile"] = yield self.profile_handler.get_profile(user_id)
res = yield self.transport_client.create_group(
get_domain_from_id(group_id), group_id, user_id, content,
)
remote_attestation = res["attestation"]
yield self.attestations.verify_attestation(
remote_attestation,
group_id=group_id,
user_id=user_id,
)
is_publicised = content.get("publicise", False)
token = yield self.store.register_user_group_membership(
group_id, user_id,
membership="join",
is_admin=True,
local_attestation=local_attestation,
remote_attestation=remote_attestation,
is_publicised=is_publicised,
)
self.notifier.on_new_event(
"groups_key", token, users=[user_id],
)
defer.returnValue(res)
@defer.inlineCallbacks
def get_users_in_group(self, group_id, requester_user_id):
"""Get users in a group
"""
if self.is_mine_id(group_id):
res = yield self.groups_server_handler.get_users_in_group(
group_id, requester_user_id
)
defer.returnValue(res)
res = yield self.transport_client.get_users_in_group(
get_domain_from_id(group_id), group_id, requester_user_id,
)
chunk = res["chunk"]
valid_entries = []
for entry in chunk:
g_user_id = entry["user_id"]
attestation = entry.pop("attestation")
try:
yield self.attestations.verify_attestation(
attestation,
group_id=group_id,
user_id=g_user_id,
)
valid_entries.append(entry)
except Exception as e:
logger.info("Failed to verify user is in group: %s", e)
res["chunk"] = valid_entries
defer.returnValue(res)
@defer.inlineCallbacks
def join_group(self, group_id, user_id, content):
"""Request to join a group
"""
raise NotImplementedError() # TODO
@defer.inlineCallbacks
def accept_invite(self, group_id, user_id, content):
"""Accept an invite to a group
"""
if self.is_mine_id(group_id):
yield self.groups_server_handler.accept_invite(
group_id, user_id, content
)
local_attestation = None
remote_attestation = None
else:
local_attestation = self.attestations.create_attestation(group_id, user_id)
content["attestation"] = local_attestation
res = yield self.transport_client.accept_group_invite(
get_domain_from_id(group_id), group_id, user_id, content,
)
remote_attestation = res["attestation"]
yield self.attestations.verify_attestation(
remote_attestation,
group_id=group_id,
user_id=user_id,
)
# TODO: Check that the group is public and we're being added publically
is_publicised = content.get("publicise", False)
token = yield self.store.register_user_group_membership(
group_id, user_id,
membership="join",
is_admin=False,
local_attestation=local_attestation,
remote_attestation=remote_attestation,
is_publicised=is_publicised,
)
self.notifier.on_new_event(
"groups_key", token, users=[user_id],
)
defer.returnValue({})
@defer.inlineCallbacks
def invite(self, group_id, user_id, requester_user_id, config):
"""Invite a user to a group
"""
content = {
"requester_user_id": requester_user_id,
"config": config,
}
if self.is_mine_id(group_id):
res = yield self.groups_server_handler.invite_to_group(
group_id, user_id, requester_user_id, content,
)
else:
res = yield self.transport_client.invite_to_group(
get_domain_from_id(group_id), group_id, user_id, requester_user_id,
content,
)
defer.returnValue(res)
@defer.inlineCallbacks
def on_invite(self, group_id, user_id, content):
"""One of our users were invited to a group
"""
# TODO: Support auto join and rejection
if not self.is_mine_id(user_id):
raise SynapseError(400, "User not on this server")
local_profile = {}
if "profile" in content:
if "name" in content["profile"]:
local_profile["name"] = content["profile"]["name"]
if "avatar_url" in content["profile"]:
local_profile["avatar_url"] = content["profile"]["avatar_url"]
token = yield self.store.register_user_group_membership(
group_id, user_id,
membership="invite",
content={"profile": local_profile, "inviter": content["inviter"]},
)
self.notifier.on_new_event(
"groups_key", token, users=[user_id],
)
user_profile = yield self.profile_handler.get_profile(user_id)
defer.returnValue({"state": "invite", "user_profile": user_profile})
@defer.inlineCallbacks
def remove_user_from_group(self, group_id, user_id, requester_user_id, content):
"""Remove a user from a group
"""
if user_id == requester_user_id:
token = yield self.store.register_user_group_membership(
group_id, user_id,
membership="leave",
)
self.notifier.on_new_event(
"groups_key", token, users=[user_id],
)
# TODO: Should probably remember that we tried to leave so that we can
# retry if the group server is currently down.
if self.is_mine_id(group_id):
res = yield self.groups_server_handler.remove_user_from_group(
group_id, user_id, requester_user_id, content,
)
else:
content["requester_user_id"] = requester_user_id
res = yield self.transport_client.remove_user_from_group(
get_domain_from_id(group_id), group_id, requester_user_id,
user_id, content,
)
defer.returnValue(res)
@defer.inlineCallbacks
def user_removed_from_group(self, group_id, user_id, content):
"""One of our users was removed/kicked from a group
"""
# TODO: Check if user in group
token = yield self.store.register_user_group_membership(
group_id, user_id,
membership="leave",
)
self.notifier.on_new_event(
"groups_key", token, users=[user_id],
)
@defer.inlineCallbacks
def get_joined_groups(self, user_id):
group_ids = yield self.store.get_joined_groups(user_id)
defer.returnValue({"groups": group_ids})
@defer.inlineCallbacks
def get_publicised_groups_for_user(self, user_id):
if self.hs.is_mine_id(user_id):
result = yield self.store.get_publicised_groups_for_user(user_id)
defer.returnValue({"groups": result})
else:
result = yield self.transport_client.get_publicised_groups_for_user(
get_domain_from_id(user_id), user_id
)
# TODO: Verify attestations
defer.returnValue(result)
@defer.inlineCallbacks
def bulk_get_publicised_groups(self, user_ids, proxy=True):
destinations = {}
local_users = set()
for user_id in user_ids:
if self.hs.is_mine_id(user_id):
local_users.add(user_id)
else:
destinations.setdefault(
get_domain_from_id(user_id), set()
).add(user_id)
if not proxy and destinations:
raise SynapseError(400, "Some user_ids are not local")
results = {}
failed_results = []
for destination, dest_user_ids in destinations.iteritems():
try:
r = yield self.transport_client.bulk_get_publicised_groups(
destination, list(dest_user_ids),
)
results.update(r["users"])
except Exception:
failed_results.extend(dest_user_ids)
for uid in local_users:
results[uid] = yield self.store.get_publicised_groups_for_user(
uid
)
defer.returnValue({"users": results})

3
synapse/handlers/message.py
View File

@ -47,6 +47,7 @@ class MessageHandler(BaseHandler):
self.state = hs.get_state_handler()
self.clock = hs.get_clock()
self.validator = EventValidator()
self.profile_handler = hs.get_profile_handler()
self.pagination_lock = ReadWriteLock()
@ -212,7 +213,7 @@ class MessageHandler(BaseHandler):
if membership in {Membership.JOIN, Membership.INVITE}:
# If event doesn't include a display name, add one.
profile = self.hs.get_handlers().profile_handler
profile = self.profile_handler
content = builder.content
try:

103
synapse/handlers/profile.py
View File

@ -19,14 +19,15 @@ from twisted.internet import defer
import synapse.types
from synapse.api.errors import SynapseError, AuthError, CodeMessageException
from synapse.types import UserID
from synapse.types import UserID, get_domain_from_id
from ._base import BaseHandler
logger = logging.getLogger(__name__)
class ProfileHandler(BaseHandler):
PROFILE_UPDATE_MS = 60 * 1000
PROFILE_UPDATE_EVERY_MS = 24 * 60 * 60 * 1000
def __init__(self, hs):
super(ProfileHandler, self).__init__(hs)
@ -36,6 +37,63 @@ class ProfileHandler(BaseHandler):
"profile", self.on_profile_query
)
self.clock.looping_call(self._update_remote_profile_cache, self.PROFILE_UPDATE_MS)
@defer.inlineCallbacks
def get_profile(self, user_id):
target_user = UserID.from_string(user_id)
if self.hs.is_mine(target_user):
displayname = yield self.store.get_profile_displayname(
target_user.localpart
)
avatar_url = yield self.store.get_profile_avatar_url(
target_user.localpart
)
defer.returnValue({
"displayname": displayname,
"avatar_url": avatar_url,
})
else:
try:
result = yield self.federation.make_query(
destination=target_user.domain,
query_type="profile",
args={
"user_id": user_id,
},
ignore_backoff=True,
)
defer.returnValue(result)
except CodeMessageException as e:
if e.code != 404:
logger.exception("Failed to get displayname")
raise
@defer.inlineCallbacks
def get_profile_from_cache(self, user_id):
"""Get the profile information from our local cache. If the user is
ours then the profile information will always be corect. Otherwise,
it may be out of date/missing.
"""
target_user = UserID.from_string(user_id)
if self.hs.is_mine(target_user):
displayname = yield self.store.get_profile_displayname(
target_user.localpart
)
avatar_url = yield self.store.get_profile_avatar_url(
target_user.localpart
)
defer.returnValue({
"displayname": displayname,
"avatar_url": avatar_url,
})
else:
profile = yield self.store.get_from_remote_profile_cache(user_id)
defer.returnValue(profile or {})
@defer.inlineCallbacks
def get_displayname(self, target_user):
if self.hs.is_mine(target_user):
@ -182,3 +240,44 @@ class ProfileHandler(BaseHandler):
"Failed to update join event for room %s - %s",
room_id, str(e.message)
)
def _update_remote_profile_cache(self):
"""Called periodically to check profiles of remote users we haven't
checked in a while.
"""
entries = yield self.store.get_remote_profile_cache_entries_that_expire(
last_checked=self.clock.time_msec() - self.PROFILE_UPDATE_EVERY_MS
)
for user_id, displayname, avatar_url in entries:
is_subscribed = yield self.store.is_subscribed_remote_profile_for_user(
user_id,
)
if not is_subscribed:
yield self.store.maybe_delete_remote_profile_cache(user_id)
continue
try:
profile = yield self.federation.make_query(
destination=get_domain_from_id(user_id),
query_type="profile",
args={
"user_id": user_id,
},
ignore_backoff=True,
)
except:
logger.exception("Failed to get avatar_url")
yield self.store.update_remote_profile_cache(
user_id, displayname, avatar_url
)
continue
new_name = profile.get("displayname")
new_avatar = profile.get("avatar_url")
# We always hit update to update the last_check timestamp
yield self.store.update_remote_profile_cache(
user_id, new_name, new_avatar
)

4
synapse/handlers/register.py
View File

@ -36,6 +36,7 @@ class RegistrationHandler(BaseHandler):
super(RegistrationHandler, self).__init__(hs)
self.auth = hs.get_auth()
self.profile_handler = hs.get_profile_handler()
self.captcha_client = CaptchaServerHttpClient(hs)
self._next_generated_user_id = None
@ -423,8 +424,7 @@ class RegistrationHandler(BaseHandler):
if displayname is not None:
logger.info("setting user display name: %s -> %s", user_id, displayname)
profile_handler = self.hs.get_handlers().profile_handler
yield profile_handler.set_displayname(
yield self.profile_handler.set_displayname(
user, requester, displayname, by_admin=True,
)

18
synapse/handlers/room_list.py
View File

@ -276,13 +276,14 @@ class RoomListHandler(BaseHandler):
# We've already got enough, so lets just drop it.
return
result = yield self._generate_room_entry(room_id, num_joined_users)
result = yield self.generate_room_entry(room_id, num_joined_users)
if result and _matches_room_entry(result, search_filter):
chunk.append(result)
@cachedInlineCallbacks(num_args=1, cache_context=True)
def _generate_room_entry(self, room_id, num_joined_users, cache_context):
def generate_room_entry(self, room_id, num_joined_users, cache_context,
with_alias=True, allow_private=False):
"""Returns the entry for a room
"""
result = {
@ -316,14 +317,15 @@ class RoomListHandler(BaseHandler):
join_rules_event = current_state.get((EventTypes.JoinRules, ""))
if join_rules_event:
join_rule = join_rules_event.content.get("join_rule", None)
if join_rule and join_rule != JoinRules.PUBLIC:
if not allow_private and join_rule and join_rule != JoinRules.PUBLIC:
defer.returnValue(None)
aliases = yield self.store.get_aliases_for_room(
room_id, on_invalidate=cache_context.invalidate
)
if aliases:
result["aliases"] = aliases
if with_alias:
aliases = yield self.store.get_aliases_for_room(
room_id, on_invalidate=cache_context.invalidate
)
if aliases:
result["aliases"] = aliases
name_event = yield current_state.get((EventTypes.Name, ""))
if name_event:

4
synapse/handlers/room_member.py
View File

@ -45,6 +45,8 @@ class RoomMemberHandler(BaseHandler):
def __init__(self, hs):
super(RoomMemberHandler, self).__init__(hs)
self.profile_handler = hs.get_profile_handler()
self.member_linearizer = Linearizer(name="member")
self.clock = hs.get_clock()
@ -282,7 +284,7 @@ class RoomMemberHandler(BaseHandler):
content["membership"] = Membership.JOIN
profile = self.hs.get_handlers().profile_handler
profile = self.profile_handler
if not content_specified:
content["displayname"] = yield profile.get_displayname(target)
content["avatar_url"] = yield profile.get_avatar_url(target)

65
synapse/handlers/sync.py
View File

@ -108,6 +108,17 @@ class InvitedSyncResult(collections.namedtuple("InvitedSyncResult", [
return True
class GroupsSyncResult(collections.namedtuple("GroupsSyncResult", [
"join",
"invite",
"leave",
])):
__slots__ = []
def __nonzero__(self):
return bool(self.join or self.invite or self.leave)
class DeviceLists(collections.namedtuple("DeviceLists", [
"changed", # list of user_ids whose devices may have changed
"left", # list of user_ids whose devices we no longer track
@ -129,6 +140,7 @@ class SyncResult(collections.namedtuple("SyncResult", [
"device_lists", # List of user_ids whose devices have chanegd
"device_one_time_keys_count", # Dict of algorithm to count for one time keys
# for this device
"groups",
])):
__slots__ = []
@ -144,7 +156,8 @@ class SyncResult(collections.namedtuple("SyncResult", [
self.archived or
self.account_data or
self.to_device or
self.device_lists
self.device_lists or
self.groups
)
@ -595,6 +608,8 @@ class SyncHandler(object):
user_id, device_id
)
yield self._generate_sync_entry_for_groups(sync_result_builder)
defer.returnValue(SyncResult(
presence=sync_result_builder.presence,
account_data=sync_result_builder.account_data,
@ -603,10 +618,57 @@ class SyncHandler(object):
archived=sync_result_builder.archived,
to_device=sync_result_builder.to_device,
device_lists=device_lists,
groups=sync_result_builder.groups,
device_one_time_keys_count=one_time_key_counts,
next_batch=sync_result_builder.now_token,
))
@measure_func("_generate_sync_entry_for_groups")
@defer.inlineCallbacks
def _generate_sync_entry_for_groups(self, sync_result_builder):
user_id = sync_result_builder.sync_config.user.to_string()
since_token = sync_result_builder.since_token
now_token = sync_result_builder.now_token
if since_token and since_token.groups_key:
results = yield self.store.get_groups_changes_for_user(
user_id, since_token.groups_key, now_token.groups_key,
)
else:
results = yield self.store.get_all_groups_for_user(
user_id, now_token.groups_key,
)
invited = {}
joined = {}
left = {}
for result in results:
membership = result["membership"]
group_id = result["group_id"]
gtype = result["type"]
content = result["content"]
if membership == "join":
if gtype == "membership":
# TODO: Add profile
content.pop("membership", None)
joined[group_id] = content["content"]
else:
joined.setdefault(group_id, {})[gtype] = content
elif membership == "invite":
if gtype == "membership":
content.pop("membership", None)
invited[group_id] = content["content"]
else:
if gtype == "membership":
left[group_id] = content["content"]
sync_result_builder.groups = GroupsSyncResult(
join=joined,
invite=invited,
leave=left,
)
@measure_func("_generate_sync_entry_for_device_list")
@defer.inlineCallbacks
def _generate_sync_entry_for_device_list(self, sync_result_builder,
@ -1368,6 +1430,7 @@ class SyncResultBuilder(object):
self.invited = []
self.archived = []
self.device = []
self.groups = None
self.to_device = []

70
synapse/http/matrixfederationclient.py
View File

@ -344,7 +344,7 @@ class MatrixFederationHttpClient(object):
@defer.inlineCallbacks
def post_json(self, destination, path, data={}, long_retries=False,
timeout=None, ignore_backoff=False):
timeout=None, ignore_backoff=False, args={}):
""" Sends the specifed json data using POST
Args:
@ -380,6 +380,7 @@ class MatrixFederationHttpClient(object):
destination,
"POST",
path,
query_bytes=encode_query_args(args),
body_callback=body_callback,
headers_dict={"Content-Type": ["application/json"]},
long_retries=long_retries,
@ -424,13 +425,6 @@ class MatrixFederationHttpClient(object):
"""
logger.debug("get_json args: %s", args)
encoded_args = {}
for k, vs in args.items():
if isinstance(vs, basestring):
vs = [vs]
encoded_args[k] = [v.encode("UTF-8") for v in vs]
query_bytes = urllib.urlencode(encoded_args, True)
logger.debug("Query bytes: %s Retry DNS: %s", args, retry_on_dns_fail)
def body_callback(method, url_bytes, headers_dict):
@ -441,7 +435,7 @@ class MatrixFederationHttpClient(object):
destination,
"GET",
path,
query_bytes=query_bytes,
query_bytes=encode_query_args(args),
body_callback=body_callback,
retry_on_dns_fail=retry_on_dns_fail,
timeout=timeout,
@ -457,6 +451,52 @@ class MatrixFederationHttpClient(object):
defer.returnValue(json.loads(body))
@defer.inlineCallbacks
def delete_json(self, destination, path, long_retries=False,
timeout=None, ignore_backoff=False, args={}):
"""Send a DELETE request to the remote expecting some json response
Args:
destination (str): The remote server to send the HTTP request
to.
path (str): The HTTP path.
long_retries (bool): A boolean that indicates whether we should
retry for a short or long time.
timeout(int): How long to try (in ms) the destination for before
giving up. None indicates no timeout.
ignore_backoff (bool): true to ignore the historical backoff data and
try the request anyway.
Returns:
Deferred: Succeeds when we get a 2xx HTTP response. The result
will be the decoded JSON body.
Fails with ``HTTPRequestException`` if we get an HTTP response
code >= 300.
Fails with ``NotRetryingDestination`` if we are not yet ready
to retry this server.
"""
response = yield self._request(
destination,
"DELETE",
path,
query_bytes=encode_query_args(args),
headers_dict={"Content-Type": ["application/json"]},
long_retries=long_retries,
timeout=timeout,
ignore_backoff=ignore_backoff,
)
if 200 <= response.code < 300:
# We need to update the transactions table to say it was sent?
check_content_type_is_json(response.headers)
with logcontext.PreserveLoggingContext():
body = yield readBody(response)
defer.returnValue(json.loads(body))
@defer.inlineCallbacks
def get_file(self, destination, path, output_stream, args={},
retry_on_dns_fail=True, max_size=None,
@ -609,3 +649,15 @@ def check_content_type_is_json(headers):
raise RuntimeError(
"Content-Type not application/json: was '%s'" % c_type
)
def encode_query_args(args):
encoded_args = {}
for k, vs in args.items():
if isinstance(vs, basestring):
vs = [vs]
encoded_args[k] = [v.encode("UTF-8") for v in vs]
query_bytes = urllib.urlencode(encoded_args, True)
return query_bytes

4
synapse/http/server.py
View File

@ -145,7 +145,9 @@ def wrap_request_handler(request_handler, include_metrics=False):
"error": "Internal server error",
"errcode": Codes.UNKNOWN,
},
send_cors=True
send_cors=True,
pretty_print=_request_user_agent_is_curl(request),
version_string=self.version_string,
)
finally:
try:

54
synapse/replication/slave/storage/groups.py Normal file
View File

@ -0,0 +1,54 @@
# -*- coding: utf-8 -*-
# Copyright 2016 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 ._base import BaseSlavedStore
from ._slaved_id_tracker import SlavedIdTracker
from synapse.storage import DataStore
from synapse.util.caches.stream_change_cache import StreamChangeCache
class SlavedGroupServerStore(BaseSlavedStore):
def __init__(self, db_conn, hs):
super(SlavedGroupServerStore, self).__init__(db_conn, hs)
self.hs = hs
self._group_updates_id_gen = SlavedIdTracker(
db_conn, "local_group_updates", "stream_id",
)
self._group_updates_stream_cache = StreamChangeCache(
"_group_updates_stream_cache", self._group_updates_id_gen.get_current_token(),
)
get_groups_changes_for_user = DataStore.get_groups_changes_for_user.__func__
get_group_stream_token = DataStore.get_group_stream_token.__func__
get_all_groups_for_user = DataStore.get_all_groups_for_user.__func__
def stream_positions(self):
result = super(SlavedGroupServerStore, self).stream_positions()
result["groups"] = self._group_updates_id_gen.get_current_token()
return result
def process_replication_rows(self, stream_name, token, rows):
if stream_name == "groups":
self._group_updates_id_gen.advance(token)
for row in rows:
self._group_updates_stream_cache.entity_has_changed(
row.user_id, token
)
return super(SlavedGroupServerStore, self).process_replication_rows(
stream_name, token, rows
)

20
synapse/replication/tcp/streams.py
View File

@ -118,6 +118,12 @@ CurrentStateDeltaStreamRow = namedtuple("CurrentStateDeltaStream", (
"state_key", # str
"event_id", # str, optional
))
GroupsStreamRow = namedtuple("GroupsStreamRow", (
"group_id", # str
"user_id", # str
"type", # str
"content", # dict
))
class Stream(object):
@ -464,6 +470,19 @@ class CurrentStateDeltaStream(Stream):
super(CurrentStateDeltaStream, self).__init__(hs)
class GroupServerStream(Stream):
NAME = "groups"
ROW_TYPE = GroupsStreamRow
def __init__(self, hs):
store = hs.get_datastore()
self.current_token = store.get_group_stream_token
self.update_function = store.get_all_groups_changes
super(GroupServerStream, self).__init__(hs)
STREAMS_MAP = {
stream.NAME: stream
for stream in (
@ -482,5 +501,6 @@ STREAMS_MAP = {
TagAccountDataStream,
AccountDataStream,
CurrentStateDeltaStream,
GroupServerStream,
)
}

2
synapse/rest/__init__.py
View File

@ -52,6 +52,7 @@ from synapse.rest.client.v2_alpha import (
thirdparty,
sendtodevice,
user_directory,
groups,
)
from synapse.http.server import JsonResource
@ -102,3 +103,4 @@ class ClientRestResource(JsonResource):
thirdparty.register_servlets(hs, client_resource)
sendtodevice.register_servlets(hs, client_resource)
user_directory.register_servlets(hs, client_resource)
groups.register_servlets(hs, client_resource)

18
synapse/rest/client/v1/profile.py
View File

@ -26,13 +26,13 @@ class ProfileDisplaynameRestServlet(ClientV1RestServlet):
def __init__(self, hs):
super(ProfileDisplaynameRestServlet, self).__init__(hs)
self.handlers = hs.get_handlers()
self.profile_handler = hs.get_profile_handler()
@defer.inlineCallbacks
def on_GET(self, request, user_id):
user = UserID.from_string(user_id)
displayname = yield self.handlers.profile_handler.get_displayname(
displayname = yield self.profile_handler.get_displayname(
user,
)
@ -55,7 +55,7 @@ class ProfileDisplaynameRestServlet(ClientV1RestServlet):
except:
defer.returnValue((400, "Unable to parse name"))
yield self.handlers.profile_handler.set_displayname(
yield self.profile_handler.set_displayname(
user, requester, new_name, is_admin)
defer.returnValue((200, {}))
@ -69,13 +69,13 @@ class ProfileAvatarURLRestServlet(ClientV1RestServlet):
def __init__(self, hs):
super(ProfileAvatarURLRestServlet, self).__init__(hs)
self.handlers = hs.get_handlers()
self.profile_handler = hs.get_profile_handler()
@defer.inlineCallbacks
def on_GET(self, request, user_id):
user = UserID.from_string(user_id)
avatar_url = yield self.handlers.profile_handler.get_avatar_url(
avatar_url = yield self.profile_handler.get_avatar_url(
user,
)
@ -97,7 +97,7 @@ class ProfileAvatarURLRestServlet(ClientV1RestServlet):
except:
defer.returnValue((400, "Unable to parse name"))
yield self.handlers.profile_handler.set_avatar_url(
yield self.profile_handler.set_avatar_url(
user, requester, new_name, is_admin)
defer.returnValue((200, {}))
@ -111,16 +111,16 @@ class ProfileRestServlet(ClientV1RestServlet):
def __init__(self, hs):
super(ProfileRestServlet, self).__init__(hs)
self.handlers = hs.get_handlers()
self.profile_handler = hs.get_profile_handler()
@defer.inlineCallbacks
def on_GET(self, request, user_id):
user = UserID.from_string(user_id)
displayname = yield self.handlers.profile_handler.get_displayname(
displayname = yield self.profile_handler.get_displayname(
user,
)
avatar_url = yield self.handlers.profile_handler.get_avatar_url(
avatar_url = yield self.profile_handler.get_avatar_url(
user,
)

695
synapse/rest/client/v2_alpha/groups.py Normal file
View File

@ -0,0 +1,695 @@
# -*- coding: utf-8 -*-
# Copyright 2017 Vector Creations 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.http.servlet import RestServlet, parse_json_object_from_request
from synapse.types import GroupID
from ._base import client_v2_patterns
import logging
logger = logging.getLogger(__name__)
class GroupServlet(RestServlet):
"""Get the group profile
"""
PATTERNS = client_v2_patterns("/groups/(?P<group_id>[^/]*)/profile$")
def __init__(self, hs):
super(GroupServlet, self).__init__()
self.auth = hs.get_auth()
self.clock = hs.get_clock()
self.groups_handler = hs.get_groups_local_handler()
@defer.inlineCallbacks
def on_GET(self, request, group_id):
requester = yield self.auth.get_user_by_req(request)
user_id = requester.user.to_string()
group_description = yield self.groups_handler.get_group_profile(group_id, user_id)
defer.returnValue((200, group_description))
@defer.inlineCallbacks
def on_POST(self, request, group_id):
requester = yield self.auth.get_user_by_req(request)
user_id = requester.user.to_string()
content = parse_json_object_from_request(request)
yield self.groups_handler.update_group_profile(
group_id, user_id, content,
)
defer.returnValue((200, {}))
class GroupSummaryServlet(RestServlet):
"""Get the full group summary
"""
PATTERNS = client_v2_patterns("/groups/(?P<group_id>[^/]*)/summary$")
def __init__(self, hs):
super(GroupSummaryServlet, self).__init__()
self.auth = hs.get_auth()
self.clock = hs.get_clock()
self.groups_handler = hs.get_groups_local_handler()
@defer.inlineCallbacks
def on_GET(self, request, group_id):
requester = yield self.auth.get_user_by_req(request)
user_id = requester.user.to_string()
get_group_summary = yield self.groups_handler.get_group_summary(group_id, user_id)
defer.returnValue((200, get_group_summary))
class GroupSummaryRoomsCatServlet(RestServlet):
"""Update/delete a rooms entry in the summary.
Matches both:
- /groups/:group/summary/rooms/:room_id
- /groups/:group/summary/categories/:category/rooms/:room_id
"""
PATTERNS = client_v2_patterns(
"/groups/(?P<group_id>[^/]*)/summary"
"(/categories/(?P<category_id>[^/]+))?"
"/rooms/(?P<room_id>[^/]*)$"
)
def __init__(self, hs):
super(GroupSummaryRoomsCatServlet, self).__init__()
self.auth = hs.get_auth()
self.clock = hs.get_clock()
self.groups_handler = hs.get_groups_local_handler()
@defer.inlineCallbacks
def on_PUT(self, request, group_id, category_id, room_id):
requester = yield self.auth.get_user_by_req(request)
user_id = requester.user.to_string()
content = parse_json_object_from_request(request)
resp = yield self.groups_handler.update_group_summary_room(
group_id, user_id,
room_id=room_id,
category_id=category_id,
content=content,
)
defer.returnValue((200, resp))
@defer.inlineCallbacks
def on_DELETE(self, request, group_id, category_id, room_id):
requester = yield self.auth.get_user_by_req(request)
user_id = requester.user.to_string()
resp = yield self.groups_handler.delete_group_summary_room(
group_id, user_id,
room_id=room_id,
category_id=category_id,
)
defer.returnValue((200, resp))
class GroupCategoryServlet(RestServlet):
"""Get/add/update/delete a group category
"""
PATTERNS = client_v2_patterns(
"/groups/(?P<group_id>[^/]*)/categories/(?P<category_id>[^/]+)$"
)
def __init__(self, hs):
super(GroupCategoryServlet, self).__init__()
self.auth = hs.get_auth()
self.clock = hs.get_clock()
self.groups_handler = hs.get_groups_local_handler()
@defer.inlineCallbacks
def on_GET(self, request, group_id, category_id):
requester = yield self.auth.get_user_by_req(request)
user_id = requester.user.to_string()
category = yield self.groups_handler.get_group_category(
group_id, user_id,
category_id=category_id,
)
defer.returnValue((200, category))
@defer.inlineCallbacks
def on_PUT(self, request, group_id, category_id):
requester = yield self.auth.get_user_by_req(request)
user_id = requester.user.to_string()
content = parse_json_object_from_request(request)
resp = yield self.groups_handler.update_group_category(
group_id, user_id,
category_id=category_id,
content=content,
)
defer.returnValue((200, resp))
@defer.inlineCallbacks
def on_DELETE(self, request, group_id, category_id):
requester = yield self.auth.get_user_by_req(request)
user_id = requester.user.to_string()
resp = yield self.groups_handler.delete_group_category(
group_id, user_id,
category_id=category_id,
)
defer.returnValue((200, resp))
class GroupCategoriesServlet(RestServlet):
"""Get all group categories
"""
PATTERNS = client_v2_patterns(
"/groups/(?P<group_id>[^/]*)/categories/$"
)
def __init__(self, hs):
super(GroupCategoriesServlet, self).__init__()
self.auth = hs.get_auth()
self.clock = hs.get_clock()
self.groups_handler = hs.get_groups_local_handler()
@defer.inlineCallbacks
def on_GET(self, request, group_id):
requester = yield self.auth.get_user_by_req(request)
user_id = requester.user.to_string()
category = yield self.groups_handler.get_group_categories(
group_id, user_id,
)
defer.returnValue((200, category))
class GroupRoleServlet(RestServlet):
"""Get/add/update/delete a group role
"""
PATTERNS = client_v2_patterns(
"/groups/(?P<group_id>[^/]*)/roles/(?P<role_id>[^/]+)$"
)
def __init__(self, hs):
super(GroupRoleServlet, self).__init__()
self.auth = hs.get_auth()
self.clock = hs.get_clock()
self.groups_handler = hs.get_groups_local_handler()
@defer.inlineCallbacks
def on_GET(self, request, group_id, role_id):
requester = yield self.auth.get_user_by_req(request)
user_id = requester.user.to_string()
category = yield self.groups_handler.get_group_role(
group_id, user_id,
role_id=role_id,
)
defer.returnValue((200, category))
@defer.inlineCallbacks
def on_PUT(self, request, group_id, role_id):
requester = yield self.auth.get_user_by_req(request)
user_id = requester.user.to_string()
content = parse_json_object_from_request(request)
resp = yield self.groups_handler.update_group_role(
group_id, user_id,
role_id=role_id,
content=content,
)
defer.returnValue((200, resp))
@defer.inlineCallbacks
def on_DELETE(self, request, group_id, role_id):
requester = yield self.auth.get_user_by_req(request)
user_id = requester.user.to_string()
resp = yield self.groups_handler.delete_group_role(
group_id, user_id,
role_id=role_id,
)
defer.returnValue((200, resp))
class GroupRolesServlet(RestServlet):
"""Get all group roles
"""
PATTERNS = client_v2_patterns(
"/groups/(?P<group_id>[^/]*)/roles/$"
)
def __init__(self, hs):
super(GroupRolesServlet, self).__init__()
self.auth = hs.get_auth()
self.clock = hs.get_clock()
self.groups_handler = hs.get_groups_local_handler()
@defer.inlineCallbacks
def on_GET(self, request, group_id):
requester = yield self.auth.get_user_by_req(request)
user_id = requester.user.to_string()
category = yield self.groups_handler.get_group_roles(
group_id, user_id,
)
defer.returnValue((200, category))
class GroupSummaryUsersRoleServlet(RestServlet):
"""Update/delete a user's entry in the summary.
Matches both:
- /groups/:group/summary/users/:room_id
- /groups/:group/summary/roles/:role/users/:user_id
"""
PATTERNS = client_v2_patterns(
"/groups/(?P<group_id>[^/]*)/summary"
"(/roles/(?P<role_id>[^/]+))?"
"/users/(?P<user_id>[^/]*)$"
)
def __init__(self, hs):
super(GroupSummaryUsersRoleServlet, self).__init__()
self.auth = hs.get_auth()
self.clock = hs.get_clock()
self.groups_handler = hs.get_groups_local_handler()
@defer.inlineCallbacks
def on_PUT(self, request, group_id, role_id, user_id):
requester = yield self.auth.get_user_by_req(request)
requester_user_id = requester.user.to_string()
content = parse_json_object_from_request(request)
resp = yield self.groups_handler.update_group_summary_user(
group_id, requester_user_id,
user_id=user_id,
role_id=role_id,
content=content,
)
defer.returnValue((200, resp))
@defer.inlineCallbacks
def on_DELETE(self, request, group_id, role_id, user_id):
requester = yield self.auth.get_user_by_req(request)
requester_user_id = requester.user.to_string()
resp = yield self.groups_handler.delete_group_summary_user(
group_id, requester_user_id,
user_id=user_id,
role_id=role_id,
)
defer.returnValue((200, resp))
class GroupRoomServlet(RestServlet):
"""Get all rooms in a group
"""
PATTERNS = client_v2_patterns("/groups/(?P<group_id>[^/]*)/rooms$")
def __init__(self, hs):
super(GroupRoomServlet, self).__init__()
self.auth = hs.get_auth()
self.clock = hs.get_clock()
self.groups_handler = hs.get_groups_local_handler()
@defer.inlineCallbacks
def on_GET(self, request, group_id):
requester = yield self.auth.get_user_by_req(request)
user_id = requester.user.to_string()
result = yield self.groups_handler.get_rooms_in_group(group_id, user_id)
defer.returnValue((200, result))
class GroupUsersServlet(RestServlet):
"""Get all users in a group
"""
PATTERNS = client_v2_patterns("/groups/(?P<group_id>[^/]*)/users$")
def __init__(self, hs):
super(GroupUsersServlet, self).__init__()
self.auth = hs.get_auth()
self.clock = hs.get_clock()
self.groups_handler = hs.get_groups_local_handler()
@defer.inlineCallbacks
def on_GET(self, request, group_id):
requester = yield self.auth.get_user_by_req(request)
user_id = requester.user.to_string()
result = yield self.groups_handler.get_users_in_group(group_id, user_id)
defer.returnValue((200, result))
class GroupCreateServlet(RestServlet):
"""Create a group
"""
PATTERNS = client_v2_patterns("/create_group$")
def __init__(self, hs):
super(GroupCreateServlet, self).__init__()
self.auth = hs.get_auth()
self.clock = hs.get_clock()
self.groups_handler = hs.get_groups_local_handler()
self.server_name = hs.hostname
@defer.inlineCallbacks
def on_POST(self, request):
requester = yield self.auth.get_user_by_req(request)
user_id = requester.user.to_string()
# TODO: Create group on remote server
content = parse_json_object_from_request(request)
localpart = content.pop("localpart")
group_id = GroupID.create(localpart, self.server_name).to_string()
result = yield self.groups_handler.create_group(group_id, user_id, content)
defer.returnValue((200, result))
class GroupAdminRoomsServlet(RestServlet):
"""Add a room to the group
"""
PATTERNS = client_v2_patterns(
"/groups/(?P<group_id>[^/]*)/admin/rooms/(?P<room_id>[^/]*)$"
)
def __init__(self, hs):
super(GroupAdminRoomsServlet, self).__init__()
self.auth = hs.get_auth()
self.clock = hs.get_clock()
self.groups_handler = hs.get_groups_local_handler()
@defer.inlineCallbacks
def on_PUT(self, request, group_id, room_id):
requester = yield self.auth.get_user_by_req(request)
user_id = requester.user.to_string()
content = parse_json_object_from_request(request)
result = yield self.groups_handler.add_room_to_group(
group_id, user_id, room_id, content,
)
defer.returnValue((200, result))
@defer.inlineCallbacks
def on_DELETE(self, request, group_id, room_id):
requester = yield self.auth.get_user_by_req(request)
user_id = requester.user.to_string()
result = yield self.groups_handler.remove_room_from_group(
group_id, user_id, room_id,
)
defer.returnValue((200, result))
class GroupAdminUsersInviteServlet(RestServlet):
"""Invite a user to the group
"""
PATTERNS = client_v2_patterns(
"/groups/(?P<group_id>[^/]*)/admin/users/invite/(?P<user_id>[^/]*)$"
)
def __init__(self, hs):
super(GroupAdminUsersInviteServlet, self).__init__()
self.auth = hs.get_auth()
self.clock = hs.get_clock()
self.groups_handler = hs.get_groups_local_handler()
self.store = hs.get_datastore()
self.is_mine_id = hs.is_mine_id
@defer.inlineCallbacks
def on_PUT(self, request, group_id, user_id):
requester = yield self.auth.get_user_by_req(request)
requester_user_id = requester.user.to_string()
content = parse_json_object_from_request(request)
config = content.get("config", {})
result = yield self.groups_handler.invite(
group_id, user_id, requester_user_id, config,
)
defer.returnValue((200, result))
class GroupAdminUsersKickServlet(RestServlet):
"""Kick a user from the group
"""
PATTERNS = client_v2_patterns(
"/groups/(?P<group_id>[^/]*)/admin/users/remove/(?P<user_id>[^/]*)$"
)
def __init__(self, hs):
super(GroupAdminUsersKickServlet, self).__init__()
self.auth = hs.get_auth()
self.clock = hs.get_clock()
self.groups_handler = hs.get_groups_local_handler()
@defer.inlineCallbacks
def on_PUT(self, request, group_id, user_id):
requester = yield self.auth.get_user_by_req(request)
requester_user_id = requester.user.to_string()
content = parse_json_object_from_request(request)
result = yield self.groups_handler.remove_user_from_group(
group_id, user_id, requester_user_id, content,
)
defer.returnValue((200, result))
class GroupSelfLeaveServlet(RestServlet):
"""Leave a joined group
"""
PATTERNS = client_v2_patterns(
"/groups/(?P<group_id>[^/]*)/self/leave$"
)
def __init__(self, hs):
super(GroupSelfLeaveServlet, self).__init__()
self.auth = hs.get_auth()
self.clock = hs.get_clock()
self.groups_handler = hs.get_groups_local_handler()
@defer.inlineCallbacks
def on_PUT(self, request, group_id):
requester = yield self.auth.get_user_by_req(request)
requester_user_id = requester.user.to_string()
content = parse_json_object_from_request(request)
result = yield self.groups_handler.remove_user_from_group(
group_id, requester_user_id, requester_user_id, content,
)
defer.returnValue((200, result))
class GroupSelfJoinServlet(RestServlet):
"""Attempt to join a group, or knock
"""
PATTERNS = client_v2_patterns(
"/groups/(?P<group_id>[^/]*)/self/join$"
)
def __init__(self, hs):
super(GroupSelfJoinServlet, self).__init__()
self.auth = hs.get_auth()
self.clock = hs.get_clock()
self.groups_handler = hs.get_groups_local_handler()
@defer.inlineCallbacks
def on_PUT(self, request, group_id):
requester = yield self.auth.get_user_by_req(request)
requester_user_id = requester.user.to_string()
content = parse_json_object_from_request(request)
result = yield self.groups_handler.join_group(
group_id, requester_user_id, content,
)
defer.returnValue((200, result))
class GroupSelfAcceptInviteServlet(RestServlet):
"""Accept a group invite
"""
PATTERNS = client_v2_patterns(
"/groups/(?P<group_id>[^/]*)/self/accept_invite$"
)
def __init__(self, hs):
super(GroupSelfAcceptInviteServlet, self).__init__()
self.auth = hs.get_auth()
self.clock = hs.get_clock()
self.groups_handler = hs.get_groups_local_handler()
@defer.inlineCallbacks
def on_PUT(self, request, group_id):
requester = yield self.auth.get_user_by_req(request)
requester_user_id = requester.user.to_string()
content = parse_json_object_from_request(request)
result = yield self.groups_handler.accept_invite(
group_id, requester_user_id, content,
)
defer.returnValue((200, result))
class GroupSelfUpdatePublicityServlet(RestServlet):
"""Update whether we publicise a users membership of a group
"""
PATTERNS = client_v2_patterns(
"/groups/(?P<group_id>[^/]*)/self/update_publicity$"
)
def __init__(self, hs):
super(GroupSelfUpdatePublicityServlet, self).__init__()
self.auth = hs.get_auth()
self.clock = hs.get_clock()
self.store = hs.get_datastore()
@defer.inlineCallbacks
def on_PUT(self, request, group_id):
requester = yield self.auth.get_user_by_req(request)
requester_user_id = requester.user.to_string()
content = parse_json_object_from_request(request)
publicise = content["publicise"]
yield self.store.update_group_publicity(
group_id, requester_user_id, publicise,
)
defer.returnValue((200, {}))
class PublicisedGroupsForUserServlet(RestServlet):
"""Get the list of groups a user is advertising
"""
PATTERNS = client_v2_patterns(
"/publicised_groups/(?P<user_id>[^/]*)$"
)
def __init__(self, hs):
super(PublicisedGroupsForUserServlet, self).__init__()
self.auth = hs.get_auth()
self.clock = hs.get_clock()
self.store = hs.get_datastore()
self.groups_handler = hs.get_groups_local_handler()
@defer.inlineCallbacks
def on_GET(self, request, user_id):
yield self.auth.get_user_by_req(request)
result = yield self.groups_handler.get_publicised_groups_for_user(
user_id
)
defer.returnValue((200, result))
class PublicisedGroupsForUsersServlet(RestServlet):
"""Get the list of groups a user is advertising
"""
PATTERNS = client_v2_patterns(
"/publicised_groups$"
)
def __init__(self, hs):
super(PublicisedGroupsForUsersServlet, self).__init__()
self.auth = hs.get_auth()
self.clock = hs.get_clock()
self.store = hs.get_datastore()
self.groups_handler = hs.get_groups_local_handler()
@defer.inlineCallbacks
def on_POST(self, request):
yield self.auth.get_user_by_req(request)
content = parse_json_object_from_request(request)
user_ids = content["user_ids"]
result = yield self.groups_handler.bulk_get_publicised_groups(
user_ids
)
defer.returnValue((200, result))
class GroupsForUserServlet(RestServlet):
"""Get all groups the logged in user is joined to
"""
PATTERNS = client_v2_patterns(
"/joined_groups$"
)
def __init__(self, hs):
super(GroupsForUserServlet, self).__init__()
self.auth = hs.get_auth()
self.clock = hs.get_clock()
self.groups_handler = hs.get_groups_local_handler()
@defer.inlineCallbacks
def on_GET(self, request):
requester = yield self.auth.get_user_by_req(request)
user_id = requester.user.to_string()
result = yield self.groups_handler.get_joined_groups(user_id)
defer.returnValue((200, result))
def register_servlets(hs, http_server):
GroupServlet(hs).register(http_server)
GroupSummaryServlet(hs).register(http_server)
GroupUsersServlet(hs).register(http_server)
GroupRoomServlet(hs).register(http_server)
GroupCreateServlet(hs).register(http_server)
GroupAdminRoomsServlet(hs).register(http_server)
GroupAdminUsersInviteServlet(hs).register(http_server)
GroupAdminUsersKickServlet(hs).register(http_server)
GroupSelfLeaveServlet(hs).register(http_server)
GroupSelfJoinServlet(hs).register(http_server)
GroupSelfAcceptInviteServlet(hs).register(http_server)
GroupsForUserServlet(hs).register(http_server)
GroupCategoryServlet(hs).register(http_server)
GroupCategoriesServlet(hs).register(http_server)
GroupSummaryRoomsCatServlet(hs).register(http_server)
GroupRoleServlet(hs).register(http_server)
GroupRolesServlet(hs).register(http_server)
GroupSelfUpdatePublicityServlet(hs).register(http_server)
GroupSummaryUsersRoleServlet(hs).register(http_server)
PublicisedGroupsForUserServlet(hs).register(http_server)
PublicisedGroupsForUsersServlet(hs).register(http_server)

5
synapse/rest/client/v2_alpha/sync.py
View File

@ -200,6 +200,11 @@ class SyncRestServlet(RestServlet):
"invite": invited,
"leave": archived,
},
"groups": {
"join": sync_result.groups.join,
"invite": sync_result.groups.invite,
"leave": sync_result.groups.leave,
},
"device_one_time_keys_count": sync_result.device_one_time_keys_count,
"next_batch": sync_result.next_batch.to_string(),
}

24
synapse/server.py
View File

@ -51,6 +51,10 @@ from synapse.handlers.initial_sync import InitialSyncHandler
from synapse.handlers.receipts import ReceiptsHandler
from synapse.handlers.read_marker import ReadMarkerHandler
from synapse.handlers.user_directory import UserDirectoyHandler
from synapse.handlers.groups_local import GroupsLocalHandler
from synapse.handlers.profile import ProfileHandler
from synapse.groups.groups_server import GroupsServerHandler
from synapse.groups.attestations import GroupAttestionRenewer, GroupAttestationSigning
from synapse.http.client import SimpleHttpClient, InsecureInterceptableContextFactory
from synapse.http.matrixfederationclient import MatrixFederationHttpClient
from synapse.notifier import Notifier
@ -112,6 +116,7 @@ class HomeServer(object):
'application_service_scheduler',
'application_service_handler',
'device_message_handler',
'profile_handler',
'notifier',
'distributor',
'client_resource',
@ -140,6 +145,10 @@ class HomeServer(object):
'read_marker_handler',
'action_generator',
'user_directory_handler',
'groups_local_handler',
'groups_server_handler',
'groups_attestation_signing',
'groups_attestation_renewer',
'spam_checker',
]
@ -253,6 +262,9 @@ class HomeServer(object):
def build_initial_sync_handler(self):
return InitialSyncHandler(self)
def build_profile_handler(self):
return ProfileHandler(self)
def build_event_sources(self):
return EventSources(self)
@ -311,6 +323,18 @@ class HomeServer(object):
def build_user_directory_handler(self):
return UserDirectoyHandler(self)
def build_groups_local_handler(self):
return GroupsLocalHandler(self)
def build_groups_server_handler(self):
return GroupsServerHandler(self)
def build_groups_attestation_signing(self):
return GroupAttestationSigning(self)
def build_groups_attestation_renewer(self):
return GroupAttestionRenewer(self)
def build_spam_checker(self):
return SpamChecker(self)

18
synapse/storage/__init__.py
View File

@ -37,7 +37,7 @@ from .media_repository import MediaRepositoryStore
from .rejections import RejectionsStore
from .event_push_actions import EventPushActionsStore
from .deviceinbox import DeviceInboxStore
from .group_server import GroupServerStore
from .state import StateStore
from .signatures import SignatureStore
from .filtering import FilteringStore
@ -88,6 +88,7 @@ class DataStore(RoomMemberStore, RoomStore,
DeviceStore,
DeviceInboxStore,
UserDirectoryStore,
GroupServerStore,
):
def __init__(self, db_conn, hs):
@ -135,6 +136,9 @@ class DataStore(RoomMemberStore, RoomStore,
db_conn, "pushers", "id",
extra_tables=[("deleted_pushers", "stream_id")],
)
self._group_updates_id_gen = StreamIdGenerator(
db_conn, "local_group_updates", "stream_id",
)
if isinstance(self.database_engine, PostgresEngine):
self._cache_id_gen = StreamIdGenerator(
@ -235,6 +239,18 @@ class DataStore(RoomMemberStore, RoomStore,
prefilled_cache=curr_state_delta_prefill,
)
_group_updates_prefill, min_group_updates_id = self._get_cache_dict(
db_conn, "local_group_updates",
entity_column="user_id",
stream_column="stream_id",
max_value=self._group_updates_id_gen.get_current_token(),
limit=1000,
)
self._group_updates_stream_cache = StreamChangeCache(
"_group_updates_stream_cache", min_group_updates_id,
prefilled_cache=_group_updates_prefill,
)
cur = LoggingTransaction(
db_conn.cursor(),
name="_find_stream_orderings_for_times_txn",

51
synapse/storage/_base.py
View File

@ -743,6 +743,33 @@ class SQLBaseStore(object):
txn.execute(sql, values)
return cls.cursor_to_dict(txn)
def _simple_update(self, table, keyvalues, updatevalues, desc):
return self.runInteraction(
desc,
self._simple_update_txn,
table, keyvalues, updatevalues,
)
@staticmethod
def _simple_update_txn(txn, table, keyvalues, updatevalues):
if keyvalues:
where = "WHERE %s" % " AND ".join("%s = ?" % k for k in keyvalues.iterkeys())
else:
where = ""
update_sql = "UPDATE %s SET %s %s" % (
table,
", ".join("%s = ?" % (k,) for k in updatevalues),
where,
)
txn.execute(
update_sql,
updatevalues.values() + keyvalues.values()
)
return txn.rowcount
def _simple_update_one(self, table, keyvalues, updatevalues,
desc="_simple_update_one"):
"""Executes an UPDATE query on the named table, setting new values for
@ -768,27 +795,13 @@ class SQLBaseStore(object):
table, keyvalues, updatevalues,
)
@staticmethod
def _simple_update_one_txn(txn, table, keyvalues, updatevalues):
if keyvalues:
where = "WHERE %s" % " AND ".join("%s = ?" % k for k in keyvalues.iterkeys())
else:
where = ""
@classmethod
def _simple_update_one_txn(cls, txn, table, keyvalues, updatevalues):
rowcount = cls._simple_update_txn(txn, table, keyvalues, updatevalues)
update_sql = "UPDATE %s SET %s %s" % (
table,
", ".join("%s = ?" % (k,) for k in updatevalues),
where,
)
txn.execute(
update_sql,
updatevalues.values() + keyvalues.values()
)
if txn.rowcount == 0:
if rowcount == 0:
raise StoreError(404, "No row found")
if txn.rowcount > 1:
if rowcount > 1:
raise StoreError(500, "More than one row matched")
@staticmethod

1187
synapse/storage/group_server.py Normal file
View File

File diff suppressed because it is too large Load Diff

98
synapse/storage/profile.py
View File

@ -13,6 +13,8 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from twisted.internet import defer
from ._base import SQLBaseStore
@ -55,3 +57,99 @@ class ProfileStore(SQLBaseStore):
updatevalues={"avatar_url": new_avatar_url},
desc="set_profile_avatar_url",
)
def get_from_remote_profile_cache(self, user_id):
return self._simple_select_one(
table="remote_profile_cache",
keyvalues={"user_id": user_id},
retcols=("displayname", "avatar_url",),
allow_none=True,
desc="get_from_remote_profile_cache",
)
def add_remote_profile_cache(self, user_id, displayname, avatar_url):
"""Ensure we are caching the remote user's profiles.
This should only be called when `is_subscribed_remote_profile_for_user`
would return true for the user.
"""
return self._simple_upsert(
table="remote_profile_cache",
keyvalues={"user_id": user_id},
values={
"displayname": displayname,
"avatar_url": avatar_url,
"last_check": self._clock.time_msec(),
},
desc="add_remote_profile_cache",
)
def update_remote_profile_cache(self, user_id, displayname, avatar_url):
return self._simple_update(
table="remote_profile_cache",
keyvalues={"user_id": user_id},
values={
"displayname": displayname,
"avatar_url": avatar_url,
"last_check": self._clock.time_msec(),
},
desc="update_remote_profile_cache",
)
@defer.inlineCallbacks
def maybe_delete_remote_profile_cache(self, user_id):
"""Check if we still care about the remote user's profile, and if we
don't then remove their profile from the cache
"""
subscribed = yield self.is_subscribed_remote_profile_for_user(user_id)
if not subscribed:
yield self._simple_delete(
table="remote_profile_cache",
keyvalues={"user_id": user_id},
desc="delete_remote_profile_cache",
)
def get_remote_profile_cache_entries_that_expire(self, last_checked):
"""Get all users who haven't been checked since `last_checked`
"""
def _get_remote_profile_cache_entries_that_expire_txn(txn):
sql = """
SELECT user_id, displayname, avatar_url
FROM remote_profile_cache
WHERE last_check < ?
"""
txn.execute(sql, (last_checked,))
return self.cursor_to_dict(txn)
return self.runInteraction(
"get_remote_profile_cache_entries_that_expire",
_get_remote_profile_cache_entries_that_expire_txn,
)
@defer.inlineCallbacks
def is_subscribed_remote_profile_for_user(self, user_id):
"""Check whether we are interested in a remote user's profile.
"""
res = yield self._simple_select_one_onecol(
table="group_users",
keyvalues={"user_id": user_id},
retcol="user_id",
allow_none=True,
desc="should_update_remote_profile_cache_for_user",
)
if res:
defer.returnValue(True)
res = yield self._simple_select_one_onecol(
table="group_invites",
keyvalues={"user_id": user_id},
retcol="user_id",
allow_none=True,
desc="should_update_remote_profile_cache_for_user",
)
if res:
defer.returnValue(True)

167
synapse/storage/schema/delta/43/group_server.sql Normal file
View File

@ -0,0 +1,167 @@
/* Copyright 2017 Vector Creations 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 groups (
group_id TEXT NOT NULL,
name TEXT, -- the display name of the room
avatar_url TEXT,
short_description TEXT,
long_description TEXT
);
CREATE UNIQUE INDEX groups_idx ON groups(group_id);
-- list of users the group server thinks are joined
CREATE TABLE group_users (
group_id TEXT NOT NULL,
user_id TEXT NOT NULL,
is_admin BOOLEAN NOT NULL,
is_public BOOLEAN NOT NULL -- whether the users membership can be seen by everyone
);
CREATE INDEX groups_users_g_idx ON group_users(group_id, user_id);
CREATE INDEX groups_users_u_idx ON group_users(user_id);
-- list of users the group server thinks are invited
CREATE TABLE group_invites (
group_id TEXT NOT NULL,
user_id TEXT NOT NULL
);
CREATE INDEX groups_invites_g_idx ON group_invites(group_id, user_id);
CREATE INDEX groups_invites_u_idx ON group_invites(user_id);
CREATE TABLE group_rooms (
group_id TEXT NOT NULL,
room_id TEXT NOT NULL,
is_public BOOLEAN NOT NULL -- whether the room can be seen by everyone
);
CREATE UNIQUE INDEX groups_rooms_g_idx ON group_rooms(group_id, room_id);
CREATE INDEX groups_rooms_r_idx ON group_rooms(room_id);
-- Rooms to include in the summary
CREATE TABLE group_summary_rooms (
group_id TEXT NOT NULL,
room_id TEXT NOT NULL,
category_id TEXT NOT NULL,
room_order BIGINT NOT NULL,
is_public BOOLEAN NOT NULL, -- whether the room should be show to everyone
UNIQUE (group_id, category_id, room_id, room_order),
CHECK (room_order > 0)
);
CREATE UNIQUE INDEX group_summary_rooms_g_idx ON group_summary_rooms(group_id, room_id, category_id);
-- Categories to include in the summary
CREATE TABLE group_summary_room_categories (
group_id TEXT NOT NULL,
category_id TEXT NOT NULL,
cat_order BIGINT NOT NULL,
UNIQUE (group_id, category_id, cat_order),
CHECK (cat_order > 0)
);
-- The categories in the group
CREATE TABLE group_room_categories (
group_id TEXT NOT NULL,
category_id TEXT NOT NULL,
profile TEXT NOT NULL,
is_public BOOLEAN NOT NULL, -- whether the category should be show to everyone
UNIQUE (group_id, category_id)
);
-- The users to include in the group summary
CREATE TABLE group_summary_users (
group_id TEXT NOT NULL,
user_id TEXT NOT NULL,
role_id TEXT NOT NULL,
user_order BIGINT NOT NULL,
is_public BOOLEAN NOT NULL -- whether the user should be show to everyone
);
CREATE INDEX group_summary_users_g_idx ON group_summary_users(group_id);
-- The roles to include in the group summary
CREATE TABLE group_summary_roles (
group_id TEXT NOT NULL,
role_id TEXT NOT NULL,
role_order BIGINT NOT NULL,
UNIQUE (group_id, role_id, role_order),
CHECK (role_order > 0)
);
-- The roles in a groups
CREATE TABLE group_roles (
group_id TEXT NOT NULL,
role_id TEXT NOT NULL,
profile TEXT NOT NULL,
is_public BOOLEAN NOT NULL, -- whether the role should be show to everyone
UNIQUE (group_id, role_id)
);
-- List of attestations we've given out and need to renew
CREATE TABLE group_attestations_renewals (
group_id TEXT NOT NULL,
user_id TEXT NOT NULL,
valid_until_ms BIGINT NOT NULL
);
CREATE INDEX group_attestations_renewals_g_idx ON group_attestations_renewals(group_id, user_id);
CREATE INDEX group_attestations_renewals_u_idx ON group_attestations_renewals(user_id);
CREATE INDEX group_attestations_renewals_v_idx ON group_attestations_renewals(valid_until_ms);
-- List of attestations we've received from remotes and are interested in.
CREATE TABLE group_attestations_remote (
group_id TEXT NOT NULL,
user_id TEXT NOT NULL,
valid_until_ms BIGINT NOT NULL,
attestation_json TEXT NOT NULL
);
CREATE INDEX group_attestations_remote_g_idx ON group_attestations_remote(group_id, user_id);
CREATE INDEX group_attestations_remote_u_idx ON group_attestations_remote(user_id);
CREATE INDEX group_attestations_remote_v_idx ON group_attestations_remote(valid_until_ms);
-- The group membership for the HS's users
CREATE TABLE local_group_membership (
group_id TEXT NOT NULL,
user_id TEXT NOT NULL,
is_admin BOOLEAN NOT NULL,
membership TEXT NOT NULL,
is_publicised BOOLEAN NOT NULL, -- if the user is publicising their membership
content TEXT NOT NULL
);
CREATE INDEX local_group_membership_u_idx ON local_group_membership(user_id, group_id);
CREATE INDEX local_group_membership_g_idx ON local_group_membership(group_id);
CREATE TABLE local_group_updates (
stream_id BIGINT NOT NULL,
group_id TEXT NOT NULL,
user_id TEXT NOT NULL,
type TEXT NOT NULL,
content TEXT NOT NULL
);

28
synapse/storage/schema/delta/43/profile_cache.sql Normal file
View File

@ -0,0 +1,28 @@
/* Copyright 2017 New Vector 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.
*/
-- A subset of remote users whose profiles we have cached.
-- Whether a user is in this table or not is defined by the storage function
-- `is_subscribed_remote_profile_for_user`
CREATE TABLE remote_profile_cache (
user_id TEXT NOT NULL,
displayname TEXT,
avatar_url TEXT,
last_check BIGINT NOT NULL
);
CREATE UNIQUE INDEX remote_profile_cache_user_id ON remote_profile_cache(user_id);
CREATE INDEX remote_profile_cache_time ON remote_profile_cache(last_check);

4
synapse/streams/events.py
View File

@ -45,6 +45,7 @@ class EventSources(object):
push_rules_key, _ = self.store.get_push_rules_stream_token()
to_device_key = self.store.get_to_device_stream_token()
device_list_key = self.store.get_device_stream_token()
groups_key = self.store.get_group_stream_token()
token = StreamToken(
room_key=(
@ -65,6 +66,7 @@ class EventSources(object):
push_rules_key=push_rules_key,
to_device_key=to_device_key,
device_list_key=device_list_key,
groups_key=groups_key,
)
defer.returnValue(token)
@ -73,6 +75,7 @@ class EventSources(object):
push_rules_key, _ = self.store.get_push_rules_stream_token()
to_device_key = self.store.get_to_device_stream_token()
device_list_key = self.store.get_device_stream_token()
groups_key = self.store.get_group_stream_token()
token = StreamToken(
room_key=(
@ -93,5 +96,6 @@ class EventSources(object):
push_rules_key=push_rules_key,
to_device_key=to_device_key,
device_list_key=device_list_key,
groups_key=groups_key,
)
defer.returnValue(token)

7
synapse/types.py
View File

@ -156,6 +156,11 @@ class EventID(DomainSpecificString):
SIGIL = "$"
class GroupID(DomainSpecificString):
"""Structure representing a group ID."""
SIGIL = "+"
class StreamToken(
namedtuple("Token", (
"room_key",
@ -166,6 +171,7 @@ class StreamToken(
"push_rules_key",
"to_device_key",
"device_list_key",
"groups_key",
))
):
_SEPARATOR = "_"
@ -204,6 +210,7 @@ class StreamToken(
or (int(other.push_rules_key) < int(self.push_rules_key))
or (int(other.to_device_key) < int(self.to_device_key))
or (int(other.device_list_key) < int(self.device_list_key))
or (int(other.groups_key) < int(self.groups_key))
)
def copy_and_advance(self, key, new_value):

4
tests/handlers/test_profile.py
View File

@ -62,8 +62,6 @@ class ProfileTestCase(unittest.TestCase):
self.ratelimiter = hs.get_ratelimiter()
self.ratelimiter.send_message.return_value = (True, 0)
hs.handlers = ProfileHandlers(hs)
self.store = hs.get_datastore()
self.frank = UserID.from_string("@1234ABCD:test")
@ -72,7 +70,7 @@ class ProfileTestCase(unittest.TestCase):
yield self.store.create_profile(self.frank.localpart)
self.handler = hs.get_handlers().profile_handler
self.handler = hs.get_profile_handler()
@defer.inlineCallbacks
def test_get_my_name(self):

5
tests/handlers/test_register.py
View File

@ -40,13 +40,14 @@ class RegistrationTestCase(unittest.TestCase):
self.hs = yield setup_test_homeserver(
handlers=None,
http_client=None,
expire_access_token=True)
expire_access_token=True,
profile_handler=Mock(),
)
self.macaroon_generator = Mock(
generate_access_token=Mock(return_value='secret'))
self.hs.get_macaroon_generator = Mock(return_value=self.macaroon_generator)
self.hs.handlers = RegistrationHandlers(self.hs)
self.handler = self.hs.get_handlers().registration_handler
self.hs.get_handlers().profile_handler = Mock()
@defer.inlineCallbacks
def test_user_is_created_and_logged_in_if_doesnt_exist(self):

3
tests/rest/client/v1/test_profile.py
View File

@ -46,6 +46,7 @@ class ProfileTestCase(unittest.TestCase):
resource_for_client=self.mock_resource,
federation=Mock(),
replication_layer=Mock(),
profile_handler=self.mock_handler
)
def _get_user_by_req(request=None, allow_guest=False):
@ -53,8 +54,6 @@ class ProfileTestCase(unittest.TestCase):
hs.get_v1auth().get_user_by_req = _get_user_by_req
hs.get_handlers().profile_handler = self.mock_handler
profile.register_servlets(hs, self.mock_resource)
@defer.inlineCallbacks

4
tests/rest/client/v1/test_rooms.py
View File

@ -1032,7 +1032,7 @@ class RoomMessageListTestCase(RestTestCase):
@defer.inlineCallbacks
def test_topo_token_is_accepted(self):
token = "t1-0_0_0_0_0_0_0_0"
token = "t1-0_0_0_0_0_0_0_0_0"
(code, response) = yield self.mock_resource.trigger_get(
"/rooms/%s/messages?access_token=x&from=%s" %
(self.room_id, token))
@ -1044,7 +1044,7 @@ class RoomMessageListTestCase(RestTestCase):
@defer.inlineCallbacks
def test_stream_token_is_accepted_for_fwd_pagianation(self):
token = "s0_0_0_0_0_0_0_0"
token = "s0_0_0_0_0_0_0_0_0"
(code, response) = yield self.mock_resource.trigger_get(
"/rooms/%s/messages?access_token=x&from=%s" %
(self.room_id, token))