|
| 1 | +"""Max introspection depth rule""" |
| 2 | + |
| 3 | +from typing import Dict, Any |
| 4 | + |
| 5 | +from ...error import GraphQLError |
| 6 | +from ...language import SKIP, FieldNode, FragmentSpreadNode, Node, VisitorAction |
| 7 | +from . import ASTValidationRule, ValidationContext |
| 8 | + |
| 9 | +__all__ = ["MaxIntrospectionDepthRule"] |
| 10 | + |
| 11 | +MAX_LIST_DEPTH = 3 |
| 12 | + |
| 13 | + |
| 14 | +class MaxIntrospectionDepthRule(ASTValidationRule): |
| 15 | + """Checks maximum introspection depth""" |
| 16 | + |
| 17 | + def __init__(self, context: ValidationContext) -> None: |
| 18 | + super().__init__(context) |
| 19 | + self._visited_fragments: Dict[str, None] = {} |
| 20 | + self._get_fragment = context.get_fragment |
| 21 | + |
| 22 | + def _check_depth(self, node: Node, depth: int = 0) -> bool: |
| 23 | + """Check whether the maximum introspection depth has been reached. |
| 24 | +
|
| 25 | + Counts the depth of list fields in "__Type" recursively |
| 26 | + and returns `True` if the limit has been reached. |
| 27 | + """ |
| 28 | + if isinstance(node, FragmentSpreadNode): |
| 29 | + visited_fragments = self._visited_fragments |
| 30 | + fragment_name = node.name.value |
| 31 | + if fragment_name in visited_fragments: |
| 32 | + # Fragment cycles are handled by `NoFragmentCyclesRule`. |
| 33 | + return False |
| 34 | + fragment = self._get_fragment(fragment_name) |
| 35 | + if not fragment: |
| 36 | + # Missing fragments checks are handled by the `KnownFragmentNamesRule`. |
| 37 | + return False |
| 38 | + |
| 39 | + # Rather than following an immutable programming pattern which has |
| 40 | + # significant memory and garbage collection overhead, we've opted to take |
| 41 | + # a mutable approach for efficiency's sake. Importantly visiting a fragment |
| 42 | + # twice is fine, so long as you don't do one visit inside the other. |
| 43 | + visited_fragments[fragment_name] = None |
| 44 | + try: |
| 45 | + return self._check_depth(fragment, depth) |
| 46 | + finally: |
| 47 | + del visited_fragments[fragment_name] |
| 48 | + |
| 49 | + if isinstance(node, FieldNode) and node.name.value in ( |
| 50 | + # check all introspection lists |
| 51 | + "fields", |
| 52 | + "interfaces", |
| 53 | + "possibleTypes", |
| 54 | + "inputFields", |
| 55 | + ): |
| 56 | + depth += 1 |
| 57 | + if depth >= MAX_LIST_DEPTH: |
| 58 | + return True |
| 59 | + |
| 60 | + # hendle fields and inline fragments |
| 61 | + try: |
| 62 | + selection_set = node.selection_set # type: ignore[attr-defined] |
| 63 | + except AttributeError: # pragma: no cover |
| 64 | + selection_set = None |
| 65 | + if selection_set: |
| 66 | + for child in selection_set.selections: |
| 67 | + if self._check_depth(child, depth): |
| 68 | + return True |
| 69 | + |
| 70 | + return False |
| 71 | + |
| 72 | + def enter_field(self, node: FieldNode, *_args: Any) -> VisitorAction: |
| 73 | + if node.name.value in ("__schema", "__type") and self._check_depth(node): |
| 74 | + self.report_error( |
| 75 | + GraphQLError( |
| 76 | + "Maximum introspection depth exceeded", |
| 77 | + [node], |
| 78 | + ) |
| 79 | + ) |
| 80 | + return SKIP |
| 81 | + return None |
0 commit comments