diff --git a/backend/helpers/provision.py b/backend/helpers/provision.py
index 0fabd55c83adb1733c58559456242c7b41e05419..95e8fee60d9f797d230e59a273dc68574a68da52 100644
--- a/backend/helpers/provision.py
+++ b/backend/helpers/provision.py
@@ -19,13 +19,14 @@ class ProvisionError(Exception):
pass
class User:
- def __init__(self, kratos_id, scim_id, displayName):
+ def __init__(self, kratos_id, scim_id, username, displayName):
self.kratos_id = kratos_id
self.scim_id = scim_id
+ self.username = username
self.displayName = displayName
def ref(self, base_url):
- return f"{base_url}Users/{self.scim_id}"
+ return f"{base_url}/Users/{self.scim_id}"
class Group:
def __init__(self, scim_id, displayName, members):
@@ -51,6 +52,89 @@ class Group:
for _, member in self.members.items():
logging.debug(f" with user {member.displayName} ({member.scim_id})")
+class ScimUser:
+ """
+ A lower-level helper class that represents the SCIM representation of a
+ user, and knows how to do the actual SCIM POST/PUT calls.
+ """
+
+ __allowed = (
+ 'app',
+ 'kratos_id',
+ 'scim_id',
+ 'active',
+ 'email',
+ 'username',
+ 'display_name',
+ 'include_name',
+ )
+
+ def __init__(self, **kwargs):
+ self.include_name = False
+ self.active = True
+ for k, v in kwargs.items():
+ assert(k in self.__class__.__allowed)
+ setattr(self, k, v)
+ # According to the SCIM RFC, the `userName` attribute is required, and
+ # it should be how to user identifies to the system.
+ # Zulip does not read the `emails` property, instead getting the
+ # email from the `userName` property.
+ if hasattr(self, 'email') and getattr(self, 'username', None) is None:
+ self.username = self.email
+
+ def _provision_data(self):
+ data = {
+ 'schemas': ['urn:ietf:params:scim:schemas:core:2.0:User'],
+ 'externalId': self.kratos_id,
+ 'active': self.active,
+ 'userName': self.username,
+ }
+ if hasattr(self, 'email'):
+ data['emails'] = [{
+ 'value': self.email,
+ 'primary': True
+ }]
+ if self.include_name:
+ data['displayName'] = self.display_name
+ data['name'] = {
+ 'formatted': self.display_name,
+ }
+ if hasattr(self, 'role'):
+ data['role'] = self.role
+
+ return data
+
+ def _scim_headers(self):
+ return {
+ 'Authorization': 'Bearer ' + self.app.scim_token
+ }
+
+ def create(self):
+ data = self._provision_data()
+ url = f"{self.app.scim_url}/Users"
+ response = requests.post(url, headers=self._scim_headers(), json=data)
+ logging.debug(f"Post SCIM user: {url} with data: {data} getting status: {response.status_code}")
+ try:
+ response_json = response.json()
+ except json.decoder.JSONDecodeError as e:
+ logging.info("SCIM result was not valid json:")
+ logging.info(response.content)
+ raise ProvisionError("App returned non-json data in SCIM user post.")
+ return response_json
+
+ def update(self):
+ data = self._provision_data()
+ url = f"{self.app.scim_url}/Users/{self.scim_id}"
+ response = requests.put(url, headers=self._scim_headers(), json=data)
+ logging.debug(f"Put SCIM user: {url} with data: {data} getting status: {response.status_code}")
+ try:
+ response_json = response.json()
+ except json.decoder.JSONDecodeError as e:
+ logging.info("SCIM result was not valid json:")
+ logging.info(response.content)
+ raise ProvisionError("App returned non-json data in SCIM user put.")
+ return response_json
+
# Read from the database which users need to be provisioned on which apps, and
# do the corresponding SCIM calls to those apps to do the actual provisioning.
class Provision:
@@ -183,6 +267,28 @@ class Provision:
if response.status_code == 204:
db.session.delete(app_role)
db.session.commit()
+ elif response.status_code == 400:
+ error_json = response.json()
+ try:
+ not_supported = error_json['detail'].startswith('DELETE operation not supported.')
+ except KeyError:
+ not_supported = False
+ if not_supported:
+ # Disable the app account as we can't delete it through
+ # SCIM.
+ logging.debug("App does not support deleting account, so disabling instead.")
+ ScimUser(
+ app=app,
+ kratos_id=app_role.user_id,
+ scim_id=existing_user.scim_id,
+ username=existing_user.username,
+ active=False,
+ ).update()
+ db.session.delete(app_role)
+ db.session.commit()
+ else:
+ logging.info(f"Error returned by SCIM deletion: {response.content}")
+ raise ProvisionError("App cannot delete user via SCIM.")
else:
logging.info(f"Error returned by SCIM deletion: {response.content}")
raise ProvisionError("App cannot delete user via SCIM.")
@@ -200,23 +306,21 @@ class Provision:
return
else:
raise
+ # Existing users that should not have access are disabled using the
+ # SCIM attribute `active`.
if app_role.role_id == Role.NO_ACCESS_ROLE_ID:
active = False
else:
active = True
logging.debug(f"Active user: {active}")
- data = {
- 'schemas': ['urn:ietf:params:scim:schemas:core:2.0:User'],
- 'externalId': app_role.user_id,
- 'active': active,
- # Zulip does not read the `emails` property, instead getting the
- # email from the `userName` property.
- 'userName': kratos_user.email,
- 'emails': [{
- 'value': kratos_user.email,
- 'primary': True
- }],
- }
+ scim_object = ScimUser(
+ app=app,
+ kratos_id=app_role.user_id,
+ active=active,
+ email=kratos_user.email,
+ # This is what at least Zulip needs. For Nextcloud we override this later on.
+ username=kratos_user.email,
+ )
# Decide whether to include the displayName in the SCIM data. Usually
# we want that, but if this is an existing user, and we think the
# displayName as stored in Stackspin has not changed since the last
@@ -231,11 +335,9 @@ class Provision:
logging.debug(f"Name changed: {name_changed}")
include_name = existing_user is None or name_changed is not None
if include_name:
+ scim_object.display_name = kratos_user.name
+ scim_object.include_name = True
logging.debug(f"Setting name in provisioning request.")
- data['displayName'] = kratos_user.name
- data['name'] = {
- 'formatted': kratos_user.name,
- }
# Now clear the `name_changed` attribute so we don't set the name
# again -- until it's changed again in Stackspin.
Provision.done_attribute(
@@ -251,40 +353,31 @@ class Provision:
# https://github.com/nextcloud/server/issues/5488
# We add the `stackspin-` prefix to make this compatible with the
# username generated by the sociallogin (SSO) Nextcloud app.
- data['userName'] = f"stackspin-{app_role.user_id}"
+ scim_object.username = f"stackspin-{app_role.user_id}"
if app.slug == 'zulip':
# Zulip does not accept an empty formatted name.
if include_name and (kratos_user.name is None or kratos_user.name == ''):
- data['name']['formatted'] = "name not set"
+ scim_object.display_name = "name not set"
# Zulip doesn't support SCIM user groups, but we can set the user
# role as a field on the user object.
if app_role.role_id == Role.ADMIN_ROLE_ID:
- data['role'] = 'owner'
+ scim_object.role = 'owner'
else:
- data['role'] = 'member'
+ scim_object.role = 'member'
# Now format the URL and make the SCIM request.
if existing_user is None:
- url = f"{app.scim_url}/Users"
- response = requests.post(url, headers=scim_headers, json=data)
- logging.debug(f"Post SCIM user: {url} with data: {data} getting status: {response.status_code}")
+ scim_result = scim_object.create()
else:
- url = f"{app.scim_url}/Users/{existing_user.scim_id}"
- response = requests.put(url, headers=scim_headers, json=data)
- logging.debug(f"Put SCIM user: {url} with data: {data} getting status: {response.status_code}")
- try:
- response_json = response.json()
- except json.decoder.JSONDecodeError as e:
- logging.info("SCIM result was not json")
- logging.info(response.content)
- raise ProvisionError("App returned non-json data in SCIM user put/post.")
- logging.debug(f"got: {response_json}")
+ scim_object.scim_id = existing_user.scim_id
+ scim_result = scim_object.update()
+ logging.debug(f"got: {scim_result}")
if existing_user is None:
# Because this is a new user for the app, we should read off its
# SCIM ID and store that in the Stackspin database.
- app_role.scim_id = response_json['id']
+ app_role.scim_id = scim_result['id']
db.session.commit()
- user = User(app_role.user_id, response_json['id'], kratos_user.name)
+ user = User(app_role.user_id, scim_result['id'], scim_result['userName'], kratos_user.name)
if app.scim_group_support:
if app_role.role_id == Role.ADMIN_ROLE_ID:
logging.debug(f"Adding user to admin group: {user.displayName} ({user.kratos_id})")
@@ -335,6 +428,9 @@ class Provision:
startIndex = startIndex + added
def _get_existing_users(self, app):
+ """
+ Get all current user accounts from the given app via SCIM.
+ """
scim_users = self._scim_list_users(app)
# Make a dictionary of the users, using their externalId as key, which
# is the kratos user ID.
@@ -391,7 +487,7 @@ class Provision:
kratos_id = kratos_user.uuid
else:
kratos_id = app_role.user_id
- users[kratos_id] = User(kratos_id, u['id'], u['displayName'])
+ users[kratos_id] = User(kratos_id, u['id'], u['userName'], u['displayName'])
return users
def _get_existing_groups(self, app):
@@ -414,7 +510,7 @@ class Provision:
members = {}
for member in group['members']:
scim_id = member['value']
- members[scim_id] = User(None, scim_id, member['display'])
+ members[scim_id] = User(None, scim_id, None, member['display'])
groups[group['id']] = Group(group['id'], group['displayName'], members)
return groups