feat: add user invitation doctype & related public methods (#33308)
* feat: add user invitation doctype & related public methods * style(user-invitation): execute formatters & add semgrep comments * refactor(user-invitation): use `is` to compare `None` values * fix(user-invitation): skip fetching `after_accept` for default app * fix(user-invitation): translate email templates * fix(user-invitaton): return pending invites from invite by email api * refactor(user-invitation): improve code quality * fix(user-invitation): translate all error messages * refactor(user-invitation): improve security & readability Improvements: - move invite expiration check to `daily_maintenance` - explicitly import all of the used packages - specify methods for all security-critical endpoints - improve error messages and give them suitable titles - remove unnecessary utility functions - make invitation key management secure - translate all of the subjects of the sent emails - use the `app_title` hook to create email titles - commit the work done after each iteration of the background invitation expiry checker - restructure code to improve readability - use `user.reset_password` to generate the target link - use clear long names to name identifiers - add document states with relevant colors (User Invitation doctype) - differ `sendmail` emails whenever possible - send an email to the invitation creator instead of the invitee after the invite has expired - remove `User Invitation Manager` role * fix(user-invitation): use valid emails to test doctype & related code * feat(user-invitation): support adding multiple roles * refactor(user-invitation): mark relevant fields `set only once` * feat(user-invitation): add `Cancelled` status * test(user-invitation): correct broken tests * test(user-invitation): form valid f-strings & run code formatter * feat(user-invitation): make doctype usable from desk * fix(user-invitation): remove delete permission from invitation doctype * feat(user-invitation): pass user inserted info to `after_accept` hook * refactor(user-invitation): improve custom action methods & errors Improvements: - trigger actions only when the invitation is in the `Pending` state - use lowercase letters to start error messages - handle cases where `user_invitation_hook` is not defined * refactor(user-invitation): remove site name from email templates * docs(user-invitation): add internal documentation * feat(user-invitation): add 'get pending' & cancel invites apis * fix(user-invitation): make invitation app specific * refactor(user-invitation): avoid mixing function programming * fix(user-invitation): make apis usable for app specific valid users * fix(user-invitation): allow app specific invites * feat(user-invitation): make list view & permission checks app specific * refactor(user-invitation): convert class methods to static when possible * feat(user-invitation): add `app_only_for` method to the doc * fix(user-invitation): f-string syntax error in `get_permission_query_conditions` * docs(user-invitation): add examples & improve the internal doc * refactor: rename method name static_ is unnecessary only_for doesn't make sense in this context when arguments are not roles * fix: Support POST request too We dont follow REST semantics 100%, anything that modifies something should ideally be doable with POST too. * chore: cap * fix: Avoid ignore_permissions as user arg --------- Co-authored-by: Ankush Menat <ankush@frappe.io>
This commit is contained in:
parent
3c4dc61b7a
commit
6d1008933f
20 changed files with 996 additions and 0 deletions
126
frappe/core/api/user_invitation.py
Normal file
126
frappe/core/api/user_invitation.py
Normal file
|
|
@ -0,0 +1,126 @@
|
|||
import frappe
|
||||
import frappe.utils
|
||||
from frappe import _
|
||||
from frappe.core.doctype.user_invitation.user_invitation import UserInvitation
|
||||
|
||||
|
||||
@frappe.whitelist(methods=["POST"])
|
||||
def invite_by_email(
|
||||
emails: str, roles: list[str], redirect_to_path: str, app_name: str = "frappe"
|
||||
) -> dict[str, list[str]]:
|
||||
UserInvitation.validate_role(app_name)
|
||||
|
||||
# validate emails
|
||||
frappe.utils.validate_email_address(emails, throw=True)
|
||||
email_list = frappe.utils.split_emails(emails)
|
||||
if not email_list:
|
||||
frappe.throw(title=_("Invalid input"), msg=_("No email addresses to invite"))
|
||||
|
||||
# get relevant data from the database
|
||||
accepted_invite_emails = frappe.db.get_all(
|
||||
"User Invitation",
|
||||
filters={"email": ["in", email_list], "status": "Accepted", "app_name": app_name},
|
||||
pluck="email",
|
||||
)
|
||||
pending_invite_emails = frappe.db.get_all(
|
||||
"User Invitation",
|
||||
filters={"email": ["in", email_list], "status": "Pending", "app_name": app_name},
|
||||
pluck="email",
|
||||
)
|
||||
|
||||
# create invitation documents
|
||||
to_invite = list(set(email_list) - set(accepted_invite_emails) - set(pending_invite_emails))
|
||||
for email in to_invite:
|
||||
frappe.get_doc(
|
||||
doctype="User Invitation",
|
||||
email=email,
|
||||
roles=[dict(role=role) for role in roles],
|
||||
app_name=app_name,
|
||||
redirect_to_path=redirect_to_path,
|
||||
).insert(ignore_permissions=True)
|
||||
|
||||
return {
|
||||
"accepted_invite_emails": accepted_invite_emails,
|
||||
"pending_invite_emails": pending_invite_emails,
|
||||
"invited_emails": to_invite,
|
||||
}
|
||||
|
||||
|
||||
@frappe.whitelist(allow_guest=True, methods=["GET"])
|
||||
def accept_invitation(key: str) -> None:
|
||||
_accept_invitation(key, False)
|
||||
|
||||
|
||||
# `app_name` is required for security
|
||||
@frappe.whitelist(methods=["PATCH", "POST"])
|
||||
def cancel_invitation(name: str, app_name: str):
|
||||
UserInvitation.validate_role(app_name)
|
||||
|
||||
if not frappe.db.exists("User Invitation", name):
|
||||
frappe.throw(title=_("Error"), msg=_("Invitation not found"))
|
||||
|
||||
invitation = frappe.get_doc("User Invitation", name)
|
||||
if invitation.app_name != app_name:
|
||||
# message is not specific enough for security
|
||||
frappe.throw(title=_("Error"), msg=_("Invitation not found"))
|
||||
|
||||
if invitation.status == "Cancelled":
|
||||
return {"cancelled_now": False}
|
||||
|
||||
if invitation.status != "Pending":
|
||||
frappe.throw(title=_("Error"), msg=_("Invitation cannot be cancelled"))
|
||||
|
||||
invitation.flags.ignore_permissions = True
|
||||
return {"cancelled_now": invitation.cancel_invite()}
|
||||
|
||||
|
||||
@frappe.whitelist(methods=["GET"])
|
||||
def get_pending_invitations(app_name: str):
|
||||
UserInvitation.validate_role(app_name)
|
||||
|
||||
pending_invitations = frappe.db.get_all(
|
||||
"User Invitation", fields=["name", "email"], filters={"status": "Pending", "app_name": app_name}
|
||||
)
|
||||
res = []
|
||||
for pending_invitation in pending_invitations:
|
||||
roles = frappe.db.get_all("User Role", fields=["role"], filters={"parent": pending_invitation.name})
|
||||
res.append(
|
||||
{
|
||||
"name": pending_invitation.name,
|
||||
"email": pending_invitation.email,
|
||||
"roles": [r.role for r in roles],
|
||||
}
|
||||
)
|
||||
return res
|
||||
|
||||
|
||||
def _accept_invitation(key: str, in_test: bool) -> None:
|
||||
# get invitation
|
||||
hashed_key = frappe.utils.sha256_hash(key)
|
||||
invitation_name = frappe.db.get_value("User Invitation", filters={"key": hashed_key})
|
||||
if not invitation_name:
|
||||
frappe.throw(title=_("Error"), msg=_("Invalid key"))
|
||||
invitation = frappe.get_doc("User Invitation", invitation_name)
|
||||
|
||||
# accept invitation
|
||||
invitation.accept(ignore_permissions=True)
|
||||
|
||||
user = frappe.get_doc("User", invitation.email)
|
||||
should_update_password = not user.last_password_reset_date and not bool(
|
||||
frappe.get_system_settings("disable_user_pass_login")
|
||||
)
|
||||
|
||||
# set redirect_to
|
||||
redirect_to = frappe.utils.get_url(invitation.get_redirect_to_path())
|
||||
if should_update_password:
|
||||
redirect_to = f"{user.reset_password()}&redirect_to=/{invitation.get_redirect_to_path()}"
|
||||
|
||||
# GET requests do not cause an implicit commit
|
||||
frappe.db.commit() # nosemgrep
|
||||
|
||||
if not in_test and not should_update_password:
|
||||
frappe.local.login_manager.login_as(invitation.email)
|
||||
|
||||
# set response
|
||||
frappe.local.response["type"] = "redirect"
|
||||
frappe.local.response["location"] = redirect_to
|
||||
0
frappe/core/doctype/user_invitation/__init__.py
Normal file
0
frappe/core/doctype/user_invitation/__init__.py
Normal file
Binary file not shown.
|
After Width: | Height: | Size: 190 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 217 KiB |
106
frappe/core/doctype/user_invitation/internal_doc/index.md
Normal file
106
frappe/core/doctype/user_invitation/internal_doc/index.md
Normal file
|
|
@ -0,0 +1,106 @@
|
|||
# User Invitation
|
||||
|
||||
## Index
|
||||
|
||||
- [Motivation](#motivation)
|
||||
- [How to use it?](#how-to-use-it)
|
||||
- [Whitelisted functions](#whitelisted-functions)
|
||||
- [`invite_by_email`](#invite_by_email)
|
||||
- [`accept_invitation`](#accept_invitation)
|
||||
- [`get_pending_invitations`](#get_pending_invitations)
|
||||
- [`cancel_invitation`](#cancel_invitation)
|
||||
- [Normal flow](#normal-flow)
|
||||
- [Important points](#important-points)
|
||||
|
||||
## Motivation
|
||||
|
||||
- Until now, there was no way to invite and create a new user based on a sent invitation that can be accepted or rejected by the invitee.
|
||||
- Due to this, custom Framework applications have to implement a user invitation flow. But most of the rules around this flow are generic enough to let Framework store all of the common logic associated with a typical user invitation flow.
|
||||
- This will help ensure consistency and prevent code duplication for custom Framework applications that need this type of feature.
|
||||
|
||||
## How to use it?
|
||||
|
||||
Define user invitation hooks in your app's `hooks.py` file. An example is shown below.
|
||||
|
||||

|
||||
|
||||
- `only_for`
|
||||
|
||||
Roles that are allowed to invite users to your app.
|
||||
|
||||
- `allowed_roles`
|
||||
|
||||
Roles that are allowed to be invited to your app.
|
||||
|
||||
- `after_accept`
|
||||
|
||||
Dot path of the function to execute after the user accepts the invitation.
|
||||
|
||||
```python
|
||||
from frappe.model.document import Document
|
||||
|
||||
def after_accept(
|
||||
invitation: Document,
|
||||
user: Document,
|
||||
user_inserted: bool
|
||||
) -> None:
|
||||
# your business logic here
|
||||
```
|
||||
|
||||
> `after_accept` is optional and should be used only if required.
|
||||
|
||||
At this point, you can start using the whitelisted functions under the `apis` section (`frappe/core/api/user_invitation.py`). For more information, read [whitelisted functions](#whitelisted-functions).
|
||||
|
||||
By default, only `System Manager`s can create a new invitation, view the list of invitations, or view more details associated with a single invitation **using the desk**. To enable users with specific roles to perform the mentioned actions, you might want to provide `create`, `read`, and `write` access to the relevant roles.
|
||||
|
||||
Example - If a user having the `Agent Manager` role should be able to use all of the user invitation features using the desk, these should be enabled:
|
||||
|
||||
- User Invitation doctype
|
||||

|
||||
|
||||
- Role doctype
|
||||

|
||||
|
||||
## Whitelisted functions
|
||||
|
||||
There are a few whitelisted functions that can be used to manage invitations. All of the whitelisted functions are in `frappe/core/api/user_invitation.py`.
|
||||
|
||||
### `invite_by_email`
|
||||
|
||||
Invite new emails to your application.
|
||||
|
||||

|
||||
|
||||
> The invited email will receive an email with a link to accept the invitation.
|
||||
|
||||
### `accept_invitation`
|
||||
|
||||
Enables invitees to accept the sent invitations.
|
||||
|
||||
> This function should not be used directly. The only reason this function is whitelisted is because the sent invitations contain a link that the invitees use to accept the invitations.
|
||||
|
||||
### `get_pending_invitations`
|
||||
|
||||
Get all of the pending invitations associated with an installed Framework application.
|
||||
|
||||

|
||||
|
||||
### `cancel_invitation`
|
||||
|
||||
Cancels a specific pending invitation associated with an installed Framework application.
|
||||
|
||||

|
||||
|
||||
## Normal flow
|
||||
|
||||
1. Invitations are created from the desk or by using the [`invite_by_email`](#invite_by_email) whitelisted function. An email is sent to the invited email with a link to accept the invitation.
|
||||
2. The app administrator or anyone able to use the desk can cancel invitations. Once an invitation is cancelled, an email is sent to the creator of the invitation.
|
||||
3. Once the invitation is accepted, a new user is created (if required) with the roles specified in the invitation and is redirected to the specified path.
|
||||
4. If the invitee doesn't accept the invitation within three days, the invitation is marked as expired by a background job that executes every day. Currently, there is no way to customize the expiration time.
|
||||
|
||||
## Important points:
|
||||
|
||||
- There can't be multiple pending invitations for the same app.
|
||||
- Once an invitation document is created from Desk, all of the fields are immutable except the `Redirect To Path` field which is mutable only when the invitation status is `Pending`.
|
||||
- To manually mark an invitation as expired, you can use the `expire` method on the invitation document.
|
||||
- To manually cancel an invitation, you can use the `cancel_invite` method on the invitation document.
|
||||
Binary file not shown.
|
After Width: | Height: | Size: 282 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 118 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 120 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 102 KiB |
254
frappe/core/doctype/user_invitation/test_user_invitation.py
Normal file
254
frappe/core/doctype/user_invitation/test_user_invitation.py
Normal file
|
|
@ -0,0 +1,254 @@
|
|||
# Copyright (c) 2025, Frappe Technologies and Contributors
|
||||
# See license.txt
|
||||
|
||||
import re
|
||||
|
||||
import frappe
|
||||
import frappe.utils
|
||||
from frappe.core.api.user_invitation import (
|
||||
_accept_invitation,
|
||||
cancel_invitation,
|
||||
get_pending_invitations,
|
||||
invite_by_email,
|
||||
)
|
||||
from frappe.core.doctype.user_invitation.user_invitation import mark_expired_invitations
|
||||
from frappe.tests import IntegrationTestCase
|
||||
|
||||
emails = [
|
||||
"test_user_invite1@example.com",
|
||||
"test_user_invite2@example.com",
|
||||
"test_user_invite3@example.com",
|
||||
"test_user_invite4@example.com",
|
||||
"test_user_invite5@example.com",
|
||||
]
|
||||
|
||||
|
||||
class IntegrationTestUserInvitation(IntegrationTestCase):
|
||||
"""
|
||||
Integration tests for UserInvitation.
|
||||
"""
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
super().setUpClass()
|
||||
user = frappe.new_doc("User")
|
||||
user.first_name = "Test"
|
||||
user.last_name = "123"
|
||||
user.email = emails[0]
|
||||
user.append_roles("System Manager")
|
||||
user.insert()
|
||||
frappe.set_user(emails[0])
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
super().tearDownClass()
|
||||
IntegrationTestUserInvitation.delete_all_invitations()
|
||||
IntegrationTestUserInvitation.delete_all_user_roles()
|
||||
frappe.db.delete("Email Queue")
|
||||
for user_email in emails:
|
||||
if frappe.db.exists("User", user_email):
|
||||
frappe.delete_doc("User", user_email)
|
||||
frappe.set_user("Administrator")
|
||||
# some of the code under test commit internally
|
||||
frappe.db.commit() # nosemgrep
|
||||
|
||||
@classmethod
|
||||
def delete_all_user_roles(cls):
|
||||
frappe.db.sql("DELETE FROM `tabUser Role`")
|
||||
|
||||
@classmethod
|
||||
def delete_all_invitations(cls):
|
||||
frappe.db.sql("DELETE FROM `tabUser Invitation`")
|
||||
|
||||
@classmethod
|
||||
def delete_invitation(cls, name: str):
|
||||
frappe.db.sql(f'DELETE FROM `tabUser Invitation` WHERE name = "{name}"')
|
||||
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
IntegrationTestUserInvitation.delete_all_invitations()
|
||||
IntegrationTestUserInvitation.delete_all_user_roles()
|
||||
frappe.db.delete("Email Queue")
|
||||
|
||||
def test_insert_invitation(self):
|
||||
invitation = self.get_dummy_invitation()
|
||||
self.assertEqual(len(self.get_email_names()), 0)
|
||||
invitation.insert()
|
||||
self.assertEqual(invitation.invited_by, frappe.session.user)
|
||||
self.assertEqual(invitation.status, "Pending")
|
||||
self.assertIsInstance(invitation.email_sent_at, str)
|
||||
self.assertIsInstance(invitation.key, str)
|
||||
self.assertIsInstance(invitation.roles, list)
|
||||
sent_emails = self.get_email_messages()
|
||||
self.assertEqual(len(sent_emails), 1)
|
||||
self.assertIn("invited", sent_emails[0].message.lower())
|
||||
|
||||
def test_update_invitation_status_to_expired(self):
|
||||
invitation = self.get_dummy_invitation()
|
||||
invitation.insert()
|
||||
self.assertEqual(len(self.get_email_names()), 1)
|
||||
invitation.expire()
|
||||
emails = self.get_email_messages(False)
|
||||
self.assertEqual(len(emails), 2)
|
||||
self.assertIn("expired", emails[0].message.lower())
|
||||
|
||||
def test_cancel_pending_invitation(self):
|
||||
invitation = self.get_dummy_invitation()
|
||||
invitation.insert()
|
||||
self.assertEqual(len(self.get_email_names(False)), 1)
|
||||
self.assertEqual(invitation.status, "Pending")
|
||||
invitation.cancel_invite()
|
||||
sent_emails = self.get_email_messages(False)
|
||||
self.assertEqual(len(sent_emails), 2)
|
||||
self.assertIn("cancelled", sent_emails[0].message.lower())
|
||||
|
||||
def test_cancel_accepted_invitation(self):
|
||||
invitation = self.get_dummy_invitation()
|
||||
invitation.insert()
|
||||
self.assertEqual(len(self.get_email_names(False)), 1)
|
||||
invitation.status = "Accepted"
|
||||
invitation.save()
|
||||
invitation.cancel_invite()
|
||||
self.assertEqual(len(self.get_email_names(False)), 1)
|
||||
|
||||
def test_cancel_expired_invitation(self):
|
||||
invitation = self.get_dummy_invitation()
|
||||
invitation.insert()
|
||||
self.assertEqual(len(self.get_email_names(False)), 1)
|
||||
invitation.expire()
|
||||
self.assertEqual(len(self.get_email_names(False)), 2)
|
||||
invitation.cancel_invite()
|
||||
self.assertEqual(len(self.get_email_names(False)), 2)
|
||||
|
||||
def test_mark_expired_invitations(self):
|
||||
invitation = self.get_dummy_invitation()
|
||||
invitation.insert()
|
||||
# the status of invitations older than 3 days should be set to expired
|
||||
invitation.db_set("creation", frappe.utils.add_days(frappe.utils.now(), -4))
|
||||
mark_expired_invitations()
|
||||
invitation.reload()
|
||||
self.assertEqual(invitation.status, "Expired")
|
||||
|
||||
def test_invite_by_email_api(self):
|
||||
accepted_invite_email = emails[1]
|
||||
invitation = frappe.get_doc(
|
||||
doctype="User Invitation",
|
||||
email=accepted_invite_email,
|
||||
roles=[dict(role="System Manager")],
|
||||
redirect_to_path="/abc",
|
||||
app_name="frappe",
|
||||
).insert()
|
||||
invitation.status = "Accepted"
|
||||
invitation.save()
|
||||
self.assertEqual(len(self.get_email_names(False)), 1)
|
||||
pending_invite_email = emails[2]
|
||||
frappe.get_doc(
|
||||
doctype="User Invitation",
|
||||
email=pending_invite_email,
|
||||
roles=[dict(role="System Manager")],
|
||||
redirect_to_path="/abc",
|
||||
app_name="frappe",
|
||||
).insert()
|
||||
self.assertEqual(len(self.get_email_names(False)), 2)
|
||||
email_to_invite = emails[3]
|
||||
res = invite_by_email(
|
||||
emails=", ".join([accepted_invite_email, pending_invite_email, email_to_invite]),
|
||||
roles=["System Manager"],
|
||||
redirect_to_path="/xyz",
|
||||
)
|
||||
self.assertSequenceEqual(res["accepted_invite_emails"], [accepted_invite_email])
|
||||
self.assertSequenceEqual(res["pending_invite_emails"], [pending_invite_email])
|
||||
self.assertSequenceEqual(res["invited_emails"], [email_to_invite])
|
||||
self.assertEqual(len(self.get_email_names(False)), 3)
|
||||
|
||||
def test_accept_invitation_api_pass_redirect(self):
|
||||
invitation = frappe.get_doc(
|
||||
doctype="User Invitation",
|
||||
email=emails[1],
|
||||
roles=[dict(role="System Manager")],
|
||||
redirect_to_path="/abc",
|
||||
app_name="frappe",
|
||||
).insert()
|
||||
self.assertEqual(len(frappe.get_all("User", filters={"email": invitation.email}, pluck="name")), 0)
|
||||
self.assertEqual(len(self.get_email_names(False)), 1)
|
||||
key = invitation._after_insert()
|
||||
self.assertEqual(len(self.get_email_names(False)), 2)
|
||||
_accept_invitation(key, True)
|
||||
res = frappe.local.response
|
||||
self.assertEqual(res.type, "redirect")
|
||||
pattern = f"^{re.escape(frappe.utils.get_url(''))}/update-password\\?key=.+&redirect_to=/abc$"
|
||||
self.assertRegex(res.location, pattern)
|
||||
user = frappe.get_doc("User", invitation.email)
|
||||
IntegrationTestUserInvitation.delete_invitation(invitation.name)
|
||||
frappe.delete_doc("User", user.name)
|
||||
|
||||
def test_accept_invitation_api_direct_redirect(self):
|
||||
invitation = frappe.get_doc(
|
||||
doctype="User Invitation",
|
||||
email=emails[1],
|
||||
roles=[dict(role="System Manager")],
|
||||
redirect_to_path="/abc",
|
||||
app_name="frappe",
|
||||
).insert()
|
||||
self.assertEqual(len(frappe.get_all("User", filters={"email": invitation.email}, pluck="name")), 0)
|
||||
original_disable_user_pass_login = frappe.get_system_settings("disable_user_pass_login")
|
||||
frappe.db.set_single_value("System Settings", "disable_user_pass_login", 1)
|
||||
self.assertEqual(len(self.get_email_names(False)), 1)
|
||||
key = invitation._after_insert()
|
||||
self.assertEqual(len(self.get_email_names(False)), 2)
|
||||
_accept_invitation(key, True)
|
||||
frappe.db.set_single_value(
|
||||
"System Settings", "disable_user_pass_login", original_disable_user_pass_login
|
||||
)
|
||||
res = frappe.local.response
|
||||
self.assertEqual(res.type, "redirect")
|
||||
pattern = f"^{re.escape(frappe.utils.get_url(''))}/abc$"
|
||||
self.assertRegex(res.location, pattern)
|
||||
user = frappe.get_doc("User", invitation.email)
|
||||
IntegrationTestUserInvitation.delete_invitation(invitation.name)
|
||||
frappe.delete_doc("User", user.name)
|
||||
|
||||
def test_get_pending_invitations_api(self):
|
||||
invitation = self.get_dummy_invitation()
|
||||
invitation.insert()
|
||||
invitation.reload()
|
||||
pending_invitations = get_pending_invitations("frappe")
|
||||
self.assertEqual(len(pending_invitations), 1)
|
||||
pending_invitation = pending_invitations[0]
|
||||
self.assertEqual(pending_invitation["name"], invitation.name)
|
||||
self.assertEqual(pending_invitation["email"], invitation.email)
|
||||
roles = pending_invitation["roles"]
|
||||
self.assertIsInstance(roles, list)
|
||||
self.assertSequenceEqual(roles, [r.role for r in invitation.roles])
|
||||
|
||||
def test_cancel_invitation_api(self):
|
||||
invitation = self.get_dummy_invitation()
|
||||
invitation.insert()
|
||||
invitation.reload()
|
||||
self.assertEqual(invitation.status, "Pending")
|
||||
self.assertEqual(len(self.get_email_names()), 1)
|
||||
res = cancel_invitation(invitation.name, "frappe")
|
||||
self.assertTrue(res["cancelled_now"])
|
||||
invitation.reload()
|
||||
self.assertEqual(invitation.status, "Cancelled")
|
||||
self.assertEqual(len(self.get_email_names()), 2)
|
||||
res = cancel_invitation(invitation.name, "frappe")
|
||||
self.assertFalse(res["cancelled_now"])
|
||||
self.assertEqual(len(self.get_email_names()), 2)
|
||||
|
||||
def get_dummy_invitation(self):
|
||||
return frappe.get_doc(
|
||||
doctype="User Invitation",
|
||||
email=emails[1],
|
||||
roles=[dict(role="System Manager")],
|
||||
redirect_to_path="/abc",
|
||||
app_name="frappe",
|
||||
)
|
||||
|
||||
def get_email_names(self, sent_only=True):
|
||||
filters = {"status": "Sent"} if sent_only else None
|
||||
return frappe.db.get_all("Email Queue", filters=filters, fields=["name"])
|
||||
|
||||
def get_email_messages(self, sent_only=True):
|
||||
filters = {"status": "Sent"} if sent_only else None
|
||||
return frappe.db.get_all("Email Queue", filters=filters, fields=["message"])
|
||||
23
frappe/core/doctype/user_invitation/user_invitation.js
Normal file
23
frappe/core/doctype/user_invitation/user_invitation.js
Normal file
|
|
@ -0,0 +1,23 @@
|
|||
// Copyright (c) 2025, Frappe Technologies and contributors
|
||||
// For license information, please see license.txt
|
||||
|
||||
frappe.ui.form.on("User Invitation", {
|
||||
refresh(frm) {
|
||||
frappe.xcall("frappe.apps.get_apps").then((r) => {
|
||||
const apps = r?.map((r) => r.name) ?? [];
|
||||
const default_app = "frappe";
|
||||
frm.set_df_property("app_name", "options", [default_app, ...apps]);
|
||||
if (!frm.doc.app_name) {
|
||||
frm.set_value("app_name", default_app);
|
||||
}
|
||||
});
|
||||
if (frm.doc.__islocal || frm.doc.status !== "Pending") {
|
||||
return;
|
||||
}
|
||||
frm.add_custom_button(__("Cancel"), () => {
|
||||
frappe.confirm(__("Are you sure you want to cancel the invitation?"), () =>
|
||||
frm.call("cancel_invite")
|
||||
);
|
||||
});
|
||||
},
|
||||
});
|
||||
143
frappe/core/doctype/user_invitation/user_invitation.json
Normal file
143
frappe/core/doctype/user_invitation/user_invitation.json
Normal file
|
|
@ -0,0 +1,143 @@
|
|||
{
|
||||
"actions": [],
|
||||
"allow_rename": 1,
|
||||
"creation": "2025-07-07 14:19:31.014655",
|
||||
"doctype": "DocType",
|
||||
"engine": "InnoDB",
|
||||
"field_order": [
|
||||
"email",
|
||||
"app_name",
|
||||
"redirect_to_path",
|
||||
"roles",
|
||||
"status",
|
||||
"invited_by",
|
||||
"key",
|
||||
"user",
|
||||
"email_sent_at",
|
||||
"accepted_at"
|
||||
],
|
||||
"fields": [
|
||||
{
|
||||
"fieldname": "email",
|
||||
"fieldtype": "Data",
|
||||
"in_list_view": 1,
|
||||
"label": "Email",
|
||||
"reqd": 1,
|
||||
"set_only_once": 1
|
||||
},
|
||||
{
|
||||
"fieldname": "invited_by",
|
||||
"fieldtype": "Link",
|
||||
"hidden": 1,
|
||||
"in_list_view": 1,
|
||||
"label": "Invited By",
|
||||
"options": "User",
|
||||
"read_only": 1
|
||||
},
|
||||
{
|
||||
"default": "Pending",
|
||||
"fieldname": "status",
|
||||
"fieldtype": "Select",
|
||||
"hidden": 1,
|
||||
"in_list_view": 1,
|
||||
"label": "Status",
|
||||
"options": "Pending\nAccepted\nExpired\nCancelled",
|
||||
"read_only": 1
|
||||
},
|
||||
{
|
||||
"fieldname": "email_sent_at",
|
||||
"fieldtype": "Datetime",
|
||||
"hidden": 1,
|
||||
"label": "Email Sent At",
|
||||
"read_only": 1
|
||||
},
|
||||
{
|
||||
"fieldname": "accepted_at",
|
||||
"fieldtype": "Datetime",
|
||||
"hidden": 1,
|
||||
"label": "Accepted At",
|
||||
"read_only": 1
|
||||
},
|
||||
{
|
||||
"fieldname": "user",
|
||||
"fieldtype": "Link",
|
||||
"hidden": 1,
|
||||
"label": "User",
|
||||
"options": "User",
|
||||
"read_only": 1
|
||||
},
|
||||
{
|
||||
"fieldname": "app_name",
|
||||
"fieldtype": "Select",
|
||||
"in_list_view": 1,
|
||||
"label": "App Name",
|
||||
"reqd": 1,
|
||||
"set_only_once": 1
|
||||
},
|
||||
{
|
||||
"fieldname": "redirect_to_path",
|
||||
"fieldtype": "Data",
|
||||
"label": "Redirect To Path",
|
||||
"read_only_depends_on": "eval:doc.status!==\"Pending\"",
|
||||
"reqd": 1
|
||||
},
|
||||
{
|
||||
"fieldname": "key",
|
||||
"fieldtype": "Data",
|
||||
"hidden": 1,
|
||||
"label": "Key",
|
||||
"read_only": 1
|
||||
},
|
||||
{
|
||||
"fieldname": "roles",
|
||||
"fieldtype": "Table MultiSelect",
|
||||
"label": "Roles",
|
||||
"options": "User Role",
|
||||
"read_only_depends_on": "eval:Boolean(doc.creation)",
|
||||
"reqd": 1
|
||||
}
|
||||
],
|
||||
"grid_page_length": 50,
|
||||
"index_web_pages_for_search": 1,
|
||||
"links": [],
|
||||
"modified": "2025-07-26 11:52:46.984800",
|
||||
"modified_by": "Administrator",
|
||||
"module": "Core",
|
||||
"name": "User Invitation",
|
||||
"owner": "Administrator",
|
||||
"permissions": [
|
||||
{
|
||||
"create": 1,
|
||||
"email": 1,
|
||||
"export": 1,
|
||||
"print": 1,
|
||||
"read": 1,
|
||||
"report": 1,
|
||||
"role": "System Manager",
|
||||
"share": 1,
|
||||
"write": 1
|
||||
}
|
||||
],
|
||||
"quick_entry": 1,
|
||||
"row_format": "Dynamic",
|
||||
"sort_field": "creation",
|
||||
"sort_order": "DESC",
|
||||
"states": [
|
||||
{
|
||||
"color": "Green",
|
||||
"title": "Accepted"
|
||||
},
|
||||
{
|
||||
"color": "Orange",
|
||||
"title": "Pending"
|
||||
},
|
||||
{
|
||||
"color": "Yellow",
|
||||
"title": "Expired"
|
||||
},
|
||||
{
|
||||
"color": "Red",
|
||||
"title": "Cancelled"
|
||||
}
|
||||
]
|
||||
}
|
||||
241
frappe/core/doctype/user_invitation/user_invitation.py
Normal file
241
frappe/core/doctype/user_invitation/user_invitation.py
Normal file
|
|
@ -0,0 +1,241 @@
|
|||
# Copyright (c) 2025, Frappe Technologies and contributors
|
||||
# For license information, please see license.txt
|
||||
|
||||
import frappe
|
||||
import frappe.utils
|
||||
from frappe import _
|
||||
from frappe.model.document import Document
|
||||
from frappe.permissions import get_roles
|
||||
|
||||
|
||||
class UserInvitation(Document):
|
||||
# begin: auto-generated types
|
||||
# This code is auto-generated. Do not modify anything in this block.
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from frappe.core.doctype.user_role.user_role import UserRole
|
||||
from frappe.types import DF
|
||||
|
||||
accepted_at: DF.Datetime | None
|
||||
app_name: DF.Literal[None]
|
||||
email: DF.Data
|
||||
email_sent_at: DF.Datetime | None
|
||||
invited_by: DF.Link | None
|
||||
key: DF.Data | None
|
||||
redirect_to_path: DF.Data
|
||||
roles: DF.TableMultiSelect[UserRole]
|
||||
status: DF.Literal["Pending", "Accepted", "Expired", "Cancelled"]
|
||||
user: DF.Link | None
|
||||
# end: auto-generated types
|
||||
|
||||
def before_insert(self):
|
||||
self._validate_invite()
|
||||
self.invited_by = frappe.session.user
|
||||
self.status = "Pending"
|
||||
|
||||
def after_insert(self):
|
||||
self._after_insert()
|
||||
|
||||
def accept(self, ignore_permissions: bool = False):
|
||||
accepted_now = self._accept()
|
||||
if not accepted_now:
|
||||
return
|
||||
user, user_inserted = self._upsert_user(ignore_permissions)
|
||||
self.save(ignore_permissions)
|
||||
user.save(ignore_permissions)
|
||||
self._run_after_accept_hooks(user, user_inserted)
|
||||
|
||||
@frappe.whitelist()
|
||||
def cancel_invite(self):
|
||||
if self.status != "Pending":
|
||||
return False
|
||||
self.status = "Cancelled"
|
||||
self.save()
|
||||
email_title = self._get_email_title()
|
||||
frappe.sendmail(
|
||||
recipients=self.email,
|
||||
subject=_("Invitation to join {0} cancelled").format(email_title),
|
||||
template="user_invitation_cancelled",
|
||||
args={"title": email_title},
|
||||
now=True,
|
||||
)
|
||||
return True
|
||||
|
||||
@frappe.whitelist()
|
||||
def expire(self):
|
||||
if self.status != "Pending":
|
||||
return
|
||||
self.status = "Expired"
|
||||
self.save()
|
||||
email_title = self._get_email_title()
|
||||
invited_by_user = frappe.get_doc("User", self.invited_by)
|
||||
frappe.sendmail(
|
||||
recipients=invited_by_user.email,
|
||||
subject=_("Invitation to join {0} expired").format(email_title),
|
||||
template="user_invitation_expired",
|
||||
args={"title": email_title},
|
||||
now=False,
|
||||
)
|
||||
|
||||
def _validate_invite(self):
|
||||
self._validate_app_name()
|
||||
self._validate_roles()
|
||||
self._validate_email()
|
||||
if frappe.db.get_value(
|
||||
"User Invitation", filters={"email": self.email, "status": "Accepted", "app_name": self.app_name}
|
||||
):
|
||||
frappe.throw(title=_("Error"), msg=_("invitation already accepted"))
|
||||
if frappe.db.get_value(
|
||||
"User Invitation", filters={"email": self.email, "status": "Pending", "app_name": self.app_name}
|
||||
):
|
||||
frappe.throw(title=_("Error"), msg=_("invitation already exists"))
|
||||
|
||||
def _after_insert(self):
|
||||
key = frappe.generate_hash()
|
||||
self.db_set("key", frappe.utils.sha256_hash(key))
|
||||
invite_link = frappe.utils.get_url(
|
||||
f"/api/method/frappe.core.api.user_invitation.accept_invitation?key={key}"
|
||||
)
|
||||
email_title = self._get_email_title()
|
||||
frappe.sendmail(
|
||||
recipients=self.email,
|
||||
subject=_("You've been invited to join {0}").format(email_title),
|
||||
template="user_invitation",
|
||||
args={"title": email_title, "invite_link": invite_link},
|
||||
now=True,
|
||||
)
|
||||
self.db_set("email_sent_at", frappe.utils.now())
|
||||
return key
|
||||
|
||||
def _accept(self):
|
||||
if self.status == "Accepted":
|
||||
return False
|
||||
if self.status == "Expired":
|
||||
frappe.throw(title=_("Error"), msg=_("Invitation is expired"))
|
||||
if self.status == "Cancelled":
|
||||
frappe.throw(title=_("Error"), msg=_("Invitation is cancelled"))
|
||||
self.status = "Accepted"
|
||||
self.accepted_at = frappe.utils.now()
|
||||
self.user = self.email
|
||||
return True
|
||||
|
||||
def _upsert_user(self, ignore_permissions: bool = False):
|
||||
user: Document | None = None
|
||||
user_inserted = False
|
||||
if frappe.db.exists("User", self.user):
|
||||
user = frappe.get_doc("User", self.user)
|
||||
else:
|
||||
user = frappe.new_doc("User")
|
||||
user.user_type = "System User"
|
||||
user.email = self.email
|
||||
user.first_name = self.email.split("@")[0].title()
|
||||
user.send_welcome_email = False
|
||||
user.insert(ignore_permissions)
|
||||
user_inserted = True
|
||||
user.append_roles(*[r.role for r in self.roles])
|
||||
return user, user_inserted
|
||||
|
||||
def _run_after_accept_hooks(self, user: Document, user_inserted: bool):
|
||||
user_invitation_hook = frappe.get_hooks("user_invitation", app_name=self.app_name)
|
||||
if not isinstance(user_invitation_hook, dict):
|
||||
return
|
||||
for dot_path in user_invitation_hook.get("after_accept") or []:
|
||||
frappe.call(dot_path, invitation=self, user=user, user_inserted=user_inserted)
|
||||
|
||||
def _get_email_title(self):
|
||||
return frappe.get_hooks("app_title", app_name=self.app_name)[0]
|
||||
|
||||
def _validate_app_name(self):
|
||||
UserInvitation.validate_app_name(self.app_name)
|
||||
|
||||
def _validate_roles(self):
|
||||
if self.app_name == "frappe":
|
||||
return
|
||||
user_invitation_hook = frappe.get_hooks("user_invitation", app_name=self.app_name)
|
||||
allowed_roles: list[str] = []
|
||||
if isinstance(user_invitation_hook, dict):
|
||||
allowed_roles = user_invitation_hook.get("allowed_roles") or []
|
||||
for r in self.roles:
|
||||
if r.role in allowed_roles:
|
||||
continue
|
||||
frappe.throw(
|
||||
title=_("Invalid role"),
|
||||
msg=_("{0} is not an allowed role for {1}").format(r.role, self.app_name),
|
||||
)
|
||||
|
||||
def _validate_email(self):
|
||||
frappe.utils.validate_email_address(self.email, throw=True)
|
||||
|
||||
def get_redirect_to_path(self):
|
||||
start_index = 1 if self.redirect_to_path.startswith("/") else 0
|
||||
return self.redirect_to_path[start_index:]
|
||||
|
||||
@staticmethod
|
||||
def validate_app_name(app_name: str):
|
||||
if app_name not in frappe.get_installed_apps():
|
||||
frappe.throw(title=_("Invalid app"), msg=_("application is not installed"))
|
||||
|
||||
@staticmethod
|
||||
def validate_role(app_name: str) -> None:
|
||||
UserInvitation.validate_app_name(app_name)
|
||||
user_invitation_hook = frappe.get_hooks("user_invitation", app_name=app_name)
|
||||
only_for: list[str] = []
|
||||
if isinstance(user_invitation_hook, dict):
|
||||
only_for = user_invitation_hook.get("only_for") or []
|
||||
if "System Manager" not in only_for:
|
||||
only_for.append("System Manager")
|
||||
frappe.only_for(only_for)
|
||||
|
||||
|
||||
def mark_expired_invitations() -> None:
|
||||
days = 3
|
||||
invitations_to_expire = frappe.db.get_all(
|
||||
"User Invitation",
|
||||
filters={"status": "Pending", "creation": ["<", frappe.utils.add_days(frappe.utils.now(), -days)]},
|
||||
)
|
||||
for invitation in invitations_to_expire:
|
||||
invitation = frappe.get_doc("User Invitation", invitation.name)
|
||||
invitation.expire()
|
||||
# to avoid losing work in case the job times out without finishing
|
||||
frappe.db.commit() # nosemgrep
|
||||
|
||||
|
||||
def get_allowed_apps(user: Document | None) -> list[str]:
|
||||
user_roles = set(get_user_roles(user))
|
||||
allowed_apps: list[str] = []
|
||||
for app in frappe.get_installed_apps():
|
||||
user_invitation_hooks = frappe.get_hooks("user_invitation", app_name=app)
|
||||
if not isinstance(user_invitation_hooks, dict):
|
||||
continue
|
||||
only_for = user_invitation_hooks.get("only_for") or []
|
||||
if set(only_for) & user_roles:
|
||||
allowed_apps.append(app)
|
||||
return allowed_apps
|
||||
|
||||
|
||||
def get_permission_query_conditions(user: Document | None) -> str | None:
|
||||
user = get_user(user)
|
||||
user_roles = get_user_roles(user)
|
||||
if "System Manager" in user_roles:
|
||||
return
|
||||
allowed_apps = get_allowed_apps(user)
|
||||
if not allowed_apps:
|
||||
return "false"
|
||||
allowed_apps_str = ", ".join([f'"{app}"' for app in allowed_apps])
|
||||
return f"`tabUser Invitation`.app_name IN ({allowed_apps_str})"
|
||||
|
||||
|
||||
def has_permission(
|
||||
doc: UserInvitation, user: Document | None = None, permission_type: str | None = None
|
||||
) -> bool:
|
||||
return permission_type != "delete" and doc.app_name in get_allowed_apps(user)
|
||||
|
||||
|
||||
def get_user_roles(user: Document | None) -> list[str]:
|
||||
return get_roles(get_user(user))
|
||||
|
||||
|
||||
def get_user(user: Document | None) -> Document:
|
||||
return user or frappe.session.user
|
||||
0
frappe/core/doctype/user_role/__init__.py
Normal file
0
frappe/core/doctype/user_role/__init__.py
Normal file
35
frappe/core/doctype/user_role/user_role.json
Normal file
35
frappe/core/doctype/user_role/user_role.json
Normal file
|
|
@ -0,0 +1,35 @@
|
|||
{
|
||||
"actions": [],
|
||||
"allow_rename": 1,
|
||||
"creation": "2025-07-17 10:56:04.746455",
|
||||
"doctype": "DocType",
|
||||
"editable_grid": 1,
|
||||
"engine": "InnoDB",
|
||||
"field_order": [
|
||||
"role"
|
||||
],
|
||||
"fields": [
|
||||
{
|
||||
"fieldname": "role",
|
||||
"fieldtype": "Link",
|
||||
"in_list_view": 1,
|
||||
"label": "Role",
|
||||
"options": "Role",
|
||||
"reqd": 1
|
||||
}
|
||||
],
|
||||
"grid_page_length": 50,
|
||||
"index_web_pages_for_search": 1,
|
||||
"istable": 1,
|
||||
"links": [],
|
||||
"modified": "2025-07-17 10:56:36.357715",
|
||||
"modified_by": "Administrator",
|
||||
"module": "Core",
|
||||
"name": "User Role",
|
||||
"owner": "Administrator",
|
||||
"permissions": [],
|
||||
"row_format": "Dynamic",
|
||||
"sort_field": "creation",
|
||||
"sort_order": "DESC",
|
||||
"states": []
|
||||
}
|
||||
23
frappe/core/doctype/user_role/user_role.py
Normal file
23
frappe/core/doctype/user_role/user_role.py
Normal file
|
|
@ -0,0 +1,23 @@
|
|||
# Copyright (c) 2025, Frappe Technologies and contributors
|
||||
# For license information, please see license.txt
|
||||
|
||||
# import frappe
|
||||
from frappe.model.document import Document
|
||||
|
||||
|
||||
class UserRole(Document):
|
||||
# begin: auto-generated types
|
||||
# This code is auto-generated. Do not modify anything in this block.
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from frappe.types import DF
|
||||
|
||||
parent: DF.Data
|
||||
parentfield: DF.Data
|
||||
parenttype: DF.Data
|
||||
role: DF.Link
|
||||
# end: auto-generated types
|
||||
|
||||
pass
|
||||
|
|
@ -110,6 +110,7 @@ permission_query_conditions = {
|
|||
"Workflow Action": "frappe.workflow.doctype.workflow_action.workflow_action.get_permission_query_conditions",
|
||||
"Prepared Report": "frappe.core.doctype.prepared_report.prepared_report.get_permission_query_condition",
|
||||
"File": "frappe.core.doctype.file.file.get_permission_query_conditions",
|
||||
"User Invitation": "frappe.core.doctype.user_invitation.user_invitation.get_permission_query_conditions",
|
||||
}
|
||||
|
||||
has_permission = {
|
||||
|
|
@ -127,6 +128,7 @@ has_permission = {
|
|||
"File": "frappe.core.doctype.file.file.has_permission",
|
||||
"Prepared Report": "frappe.core.doctype.prepared_report.prepared_report.has_permission",
|
||||
"Notification Settings": "frappe.desk.doctype.notification_settings.notification_settings.has_permission",
|
||||
"User Invitation": "frappe.core.doctype.user_invitation.user_invitation.has_permission",
|
||||
}
|
||||
|
||||
has_website_permission = {"Address": "frappe.contacts.doctype.address.address.has_website_permission"}
|
||||
|
|
@ -247,6 +249,7 @@ scheduler_events = {
|
|||
"frappe.website.doctype.personal_data_deletion_request.personal_data_deletion_request.remove_unverified_record",
|
||||
"frappe.automation.doctype.auto_repeat.auto_repeat.make_auto_repeat_entry",
|
||||
"frappe.core.doctype.log_settings.log_settings.run_log_clean_up",
|
||||
"frappe.core.doctype.user_invitation.user_invitation.mark_expired_invitations",
|
||||
],
|
||||
"weekly_long": [
|
||||
"frappe.desk.form.document_follow.send_weekly_updates",
|
||||
|
|
@ -564,3 +567,7 @@ persistent_cache_keys = [
|
|||
"rate-limit-counter-*",
|
||||
"rl:*",
|
||||
]
|
||||
|
||||
user_invitation = {
|
||||
"only_for": ["System Manager"],
|
||||
}
|
||||
|
|
|
|||
20
frappe/templates/emails/user_invitation.html
Normal file
20
frappe/templates/emails/user_invitation.html
Normal file
|
|
@ -0,0 +1,20 @@
|
|||
<span style = "display: block; margin-bottom: 1.5rem;">
|
||||
{{ _("Hello,") }}
|
||||
</span>
|
||||
<span style = "display: block; margin-bottom: 1rem;">
|
||||
{{ _("You've been invited to join {0}.").format(title) }}
|
||||
</span>
|
||||
<span style = "display: block; margin-bottom: 0.5rem;">
|
||||
{{ _("Click below to get started:") }}
|
||||
</span>
|
||||
<a
|
||||
href = "{{ invite_link }}"
|
||||
target = "_blank"
|
||||
class = "btn btn-primary"
|
||||
style = "display: block; margin-bottom: 1rem; width: fit-content"
|
||||
>
|
||||
{{ _("Accept Invitation") }}
|
||||
</a>
|
||||
<span>
|
||||
{{ _("If you have any questions, reach out to your system administrator.") }}
|
||||
</span>
|
||||
9
frappe/templates/emails/user_invitation_cancelled.html
Normal file
9
frappe/templates/emails/user_invitation_cancelled.html
Normal file
|
|
@ -0,0 +1,9 @@
|
|||
<span style = "display: block; margin-bottom: 1.5rem;">
|
||||
{{ _("Hello,") }}
|
||||
</span>
|
||||
<span style = "display: block; margin-bottom: 1rem;">
|
||||
{{ _("Your invitation to join {0} has been cancelled by the site administrator.").format(title) }}
|
||||
</span>
|
||||
<span style = "display: block;">
|
||||
{{ _("If this was a mistake or you need access again, please reach out to your team.") }}
|
||||
</span>
|
||||
9
frappe/templates/emails/user_invitation_expired.html
Normal file
9
frappe/templates/emails/user_invitation_expired.html
Normal file
|
|
@ -0,0 +1,9 @@
|
|||
<span style = "display: block; margin-bottom: 1.5rem;">
|
||||
{{ _("Hello,") }}
|
||||
</span>
|
||||
<span style = "display: block; margin-bottom: 1rem;">
|
||||
{{ _("Your invitation to join {0} has expired.").format(title) }}
|
||||
</span>
|
||||
<span style = "display: block;">
|
||||
{{ _("You can ask your team to resend the invitation if you'd still like to join.") }}
|
||||
</span>
|
||||
Loading…
Add table
Reference in a new issue