feat: Field Level Perms (#66)

* feat: Field Level Perms

* fix: keywords in field names

Author: fahimalizain

frappe_graphql/utils/cursor_pagination.py

@@ -3,6 +3,8 @@
 
 import frappe
 
+from frappe_graphql.utils.permissions import get_allowed_fieldnames_for_doctype
+
 
 class CursorPaginator(object):
     def __init__(
@@ -142,12 +144,16 @@ def get_data(self, doctype, filters, sorting_fields, sort_dir, limit):
 
         return frappe.get_list(
             doctype,
-            fields=["*", f"SUBSTR(\".{doctype}\", 2) as doctype"] + sorting_fields,
+            fields=self.get_fields_to_fetch(doctype, filters, sorting_fields),
             filters=filters,
             order_by=f"{', '.join([f'{x} {sort_dir}' for x in sorting_fields])}",
             limit_page_length=limit
         )
 
+    def get_fields_to_fetch(self, doctype, filters, sorting_fields):
+        fieldnames = get_allowed_fieldnames_for_doctype(doctype)
+        return list(set(fieldnames + [f"SUBSTR(\".{doctype}\", 2) as doctype"] + sorting_fields))
+
     def get_sort_args(self, sorting_input=None):
         sort_dir = self.default_sorting_direction if self.default_sorting_direction in (
         "asc", "desc") else "desc"

frappe_graphql/utils/permissions.py

@@ -0,0 +1,44 @@
+import frappe
+from frappe.model import default_fields, no_value_fields
+from frappe.model.meta import Meta
+
+
+def get_allowed_fieldnames_for_doctype(doctype: str, parent_doctype: str = None):
+    """
+    Gets a list of fieldnames that's allowed for the current User to
+    read on the specified doctype. This includes default_fields
+    """
+    fieldnames = list(default_fields) + ["\"{}\" as `doctype`".format(doctype)]
+    fieldnames.remove("doctype")
+
+    meta = frappe.get_meta(doctype)
+    has_access_to = _get_permlevel_read_access(meta=frappe.get_meta(parent_doctype or doctype))
+    if not has_access_to:
+        return []
+
+    for df in meta.fields:
+        if df.fieldtype in no_value_fields:
+            continue
+
+        if df.permlevel is not None and df.permlevel not in has_access_to:
+            continue
+
+        fieldnames.append(df.fieldname)
+
+    return fieldnames
+
+
+def _get_permlevel_read_access(meta: Meta):
+    ptype = "read"
+    _has_access_to = []
+    roles = frappe.get_roles()
+    for perm in meta.permissions:
+        if perm.get("role") not in roles or not perm.get(ptype):
+            continue
+
+        if perm.get("permlevel") in _has_access_to:
+            continue
+
+        _has_access_to.append(perm.get("permlevel"))
+
+    return _has_access_to

frappe_graphql/utils/resolver/child_tables.py

@@ -23,6 +23,11 @@ def setup_child_table_resolvers(schema: GraphQLSchema):
 
 
 def _child_table_resolver(obj, info: GraphQLResolveInfo, **kwargs):
+    # If the obj already has a non None value, we can return it.
+    # This happens when the resolver returns a full doc
+    if obj.get(info.field_name) is not None:
+        return obj.get(info.field_name)
+
     df = getattr(info.parent_type.fields[info.field_name], "frappe_docfield", None)
     if not df:
         return []

frappe_graphql/utils/resolver/dataloaders/child_table_loader.py

@@ -1,10 +1,11 @@
+from collections import OrderedDict
+
 import frappe
 
 from frappe_graphql.utils.execution import DataLoader
+from frappe_graphql.utils.permissions import get_allowed_fieldnames_for_doctype
 from .locals import get_loader_from_locals, set_loader_in_locals
 
-from collections import OrderedDict
-
 
 def get_child_table_loader(child_doctype: str, parent_doctype: str, parentfield: str) -> DataLoader:
     locals_key = (child_doctype, parent_doctype, parentfield)
@@ -23,10 +24,16 @@ def get_child_table_loader(child_doctype: str, parent_doctype: str, parentfield:
 
 def _get_child_table_loader_fn(child_doctype: str, parent_doctype: str, parentfield: str):
     def _inner(keys):
+        fieldnames = get_allowed_fieldnames_for_doctype(
+            doctype=child_doctype,
+            parent_doctype=parent_doctype
+        )
+
+        select_fields = ", ".join([f"`{x}`" if "`" not in x else x for x in fieldnames])
+
         rows = frappe.db.sql(f"""
-        SELECT 
-            *,
-            "{child_doctype}" as doctype
+        SELECT
+            {select_fields}
         FROM `tab{child_doctype}`
         WHERE
             parent IN %(parent_keys)s

frappe_graphql/utils/resolver/dataloaders/doctype_loader.py

@@ -3,6 +3,7 @@
 import frappe
 
 from frappe_graphql.utils.execution import DataLoader
+from frappe_graphql.utils.permissions import get_allowed_fieldnames_for_doctype
 from .locals import get_loader_from_locals, set_loader_in_locals
 
 
@@ -17,12 +18,13 @@ def get_doctype_dataloader(doctype: str) -> DataLoader:
 
 
 def _get_document_loader_fn(doctype: str):
+    fieldnames = get_allowed_fieldnames_for_doctype(doctype)
 
     def _load_documents(keys: List[str]):
         docs = frappe.get_list(
             doctype=doctype,
             filters=[["name", "IN", keys]],
-            fields=["*", f"'{doctype}' as doctype"],
+            fields=fieldnames,
             limit_page_length=len(keys) + 1
         )
 

frappe_graphql/utils/resolver/root_query.py

@@ -5,7 +5,6 @@
 
 from frappe_graphql import CursorPaginator
 
-from .dataloaders import get_doctype_dataloader
 from .utils import get_singular_doctype, get_plural_doctype
 
 
@@ -39,7 +38,9 @@ def _get_doc_resolver(obj, info: GraphQLResolveInfo, **kwargs):
     if not frappe.has_permission(doctype=dt, doc=dn):
         raise frappe.PermissionError(frappe._("No permission for {0}").format(dt + " " + dn))
 
-    return get_doctype_dataloader(dt).load(dn)
+    doc = frappe.get_doc(dt, dn)
+    doc.apply_fieldlevel_read_permissions()
+    return doc
 
 
 def _doc_cursor_resolver(obj, info: GraphQLResolveInfo, **kwargs):

frappe_graphql/utils/tests/test_permissions.py

@@ -0,0 +1,75 @@
+from unittest import TestCase
+from unittest.mock import patch
+
+import frappe
+from frappe.model import no_value_fields, default_fields
+
+from ..permissions import get_allowed_fieldnames_for_doctype
+
+
+class TestGetAllowedFieldNameForDocType(TestCase):
+    def setUp(self) -> None:
+        pass
+
+    def tearDown(self) -> None:
+        frappe.set_user("Administrator")
+
+    def test_admin_on_user(self):
+        """
+        Administrator on User doctype
+        """
+        meta = frappe.get_meta("User")
+        fieldnames = get_allowed_fieldnames_for_doctype("User")
+        self.assertCountEqual(
+            fieldnames,
+            [x.fieldname for x in meta.fields if x.fieldtype not in no_value_fields]
+            + [x for x in default_fields if x != "doctype"]
+            + ["\"{}\" as doctype".format(meta.name)]
+        )
+
+    def test_permlevels_on_user(self):
+        frappe.set_user("Guest")
+
+        # Guest is given permlevel=0 access on User DocType
+        user_meta = self._get_custom_user_meta()
+
+        with patch("frappe.get_meta") as get_meta_mock:
+            get_meta_mock.return_value = user_meta
+            fieldnames = get_allowed_fieldnames_for_doctype(user_meta.name)
+
+        self.assertCountEqual(
+            fieldnames,
+            [x.fieldname for x in user_meta.fields
+                if x.permlevel == 1 and x.fieldtype not in no_value_fields]
+            + [x for x in default_fields if x != "doctype"]
+            + ["\"{}\" as doctype".format(user_meta.name)]
+        )
+
+        # Clear meta_cache for User doctype
+        del frappe.local.meta_cache["User"]
+
+    def test_on_child_doctype(self):
+        fieldnames = get_allowed_fieldnames_for_doctype("Has Role", parent_doctype="User")
+        meta = frappe.get_meta("Has Role")
+        self.assertCountEqual(
+            fieldnames,
+            [x.fieldname for x in meta.fields if x.fieldtype not in no_value_fields]
+            + [x for x in default_fields if x != "doctype"]
+            + ["\"{}\" as doctype".format(meta.name)]
+        )
+
+    def test_on_child_doctype_with_no_parent_doctype(self):
+        fieldnames = get_allowed_fieldnames_for_doctype("Has Role")
+        self.assertEqual(fieldnames, [])
+
+    def _get_custom_user_meta(self):
+        meta = frappe.get_meta("User")
+        meta.permissions.append(dict(
+            role="Guest",
+            read=1,
+            permlevel=1
+        ))
+
+        meta.get_field("full_name").permlevel = 1
+
+        return meta