diff --git a/tests/tools/test_flake8_plugin.py b/tests/tools/test_flake8_plugin.py index e90fe4c479d0..1889845b039d 100644 --- a/tests/tools/test_flake8_plugin.py +++ b/tests/tools/test_flake8_plugin.py @@ -1987,6 +1987,31 @@ def put(self, request) -> Response[X]: assert _run_input(src, SHAPED) == [] +def test_S028_serializer_reached_through_a_lowercase_attribute_is_not_a_hand_off() -> None: + src = """\ +class E(Endpoint): + publish_status = {"PUT": ApiPublishStatus.PUBLIC} + + def put(self, request) -> Response[X]: + settings.validator(instance=x, data=request.data) + return self.get_serializer(data=request.data) +""" + assert _run_input(src, SHAPED) == [] + + +def test_S028_plain_call_taking_data_is_still_a_hand_off() -> None: + src = """\ +class E(Endpoint): + publish_status = {"PUT": ApiPublishStatus.PUBLIC} + + def put(self, request) -> Response[X]: + return installation.get_issue(external_id, data=request.data) +""" + errors = _run_input(src, SHAPED) + assert len(errors) == 1 + assert "handed to get_issue" in errors[0] + + def test_S028_class_variable_without_data_is_still_a_hand_off() -> None: src = """\ class E(Endpoint): @@ -2420,6 +2445,1086 @@ def get(self, request) -> Response[X]: assert _api_violations(src) == [] +def _summary(violations: list[ApiInputViolation]) -> list[tuple[ApiRule, int, str, str]]: + """(rule, line, scope, key) for each violation, in report order.""" + summary = [] + for violation in violations: + summary.append((violation.rule, violation.line, violation.scope, violation.key)) + return summary + + +def test_helper_method_read_is_reported_at_the_helper() -> None: + src = """\ +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + def get(self, request) -> Response[X]: + return self._status(request) + + def _status(self, request): + return request.GET.get("status") +""" + assert _summary(_api_violations(src)) == [("S026", 8, "E._status", "query:status")] + + +def test_helper_defined_below_and_module_function_are_followed() -> None: + src = """\ +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + def get(self, request) -> Response[X]: + self._one(request) + return build(request) + + def _one(self, request): + return request.GET.get("one") + + +def build(request): + return request.GET.get("two") +""" + assert _summary(_api_violations(src)) == [ + ("S026", 9, "E._one", "query:one"), + ("S026", 13, "build", "query:two"), + ] + + +def test_helpers_of_helpers_are_followed() -> None: + src = """\ +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + def get(self, request) -> Response[X]: + return self._outer(request) + + def _outer(self, request): + return self._inner(request) + + def _inner(self, request): + return request.GET.get("deep") +""" + assert _summary(_api_violations(src)) == [("S026", 11, "E._inner", "query:deep")] + + +def test_helper_serializer_is_checked_against_the_public_method_schema() -> None: + src = """\ +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + def get(self, request) -> Response[X]: + return self._validate(request) + + def _validate(self, request): + return QuerySerializer(data=request.GET) +""" + assert _summary(_api_violations(src)) == [("S025", 8, "E.get", "query:QuerySerializer")] + + +def test_helper_serializer_declared_on_the_public_method_is_silent() -> None: + src = """\ +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + @extend_schema(parameters=[QuerySerializer]) + def get(self, request) -> Response[X]: + return self._validate(request) + + def _validate(self, request): + return QuerySerializer(data=request.GET) +""" + assert _api_violations(src) == [] + + +def test_subclass_override_wins_over_the_base_method() -> None: + src = """\ +class Base(Endpoint): + def _limit(self, request): + return request.GET.get("base_limit") + + +class E(Base): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + def get(self, request) -> Response[X]: + return self._limit(request) + + def _limit(self, request): + return request.GET.get("limit") +""" + assert _summary(_api_violations(src)) == [("S026", 13, "E._limit", "query:limit")] + + +def test_inherited_helper_resolves_from_the_public_class() -> None: + src = """\ +class Base(Endpoint): + def _shared(self, request): + return self._limit(request) + + def _limit(self, request): + return request.GET.get("base_limit") + + +class E(Base): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + def get(self, request) -> Response[X]: + return self._shared(request) + + def _limit(self, request): + return request.GET.get("limit") +""" + assert _summary(_api_violations(src)) == [("S026", 16, "E._limit", "query:limit")] + + +def test_super_call_skips_the_defining_class() -> None: + src = """\ +class Base(Endpoint): + def _filters(self, request): + return request.GET.get("base") + + +class E(Base): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + def get(self, request) -> Response[X]: + return self._filters(request) + + def _filters(self, request): + super()._filters(request) + return request.GET.get("child") +""" + assert _summary(_api_violations(src)) == [ + ("S026", 14, "E._filters", "query:child"), + ("S026", 3, "Base._filters", "query:base"), + ] + + +def test_method_from_a_base_in_another_file_is_not_followed() -> None: + src = """\ +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + def get(self, request) -> Response[X]: + self.get_filters(request) + return self.paginate(request.GET) +""" + assert _summary(_api_violations(src)) == [("S028", 6, "E.get", "query:paginate")] + + +def test_hand_off_to_a_followed_helper_is_dropped_but_other_hand_offs_stay() -> None: + src = """\ +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + def get(self, request) -> Response[X]: + self._use(request.GET) + return parse_filters(request.GET) + + def _use(self, params): + return None +""" + assert _summary(_api_violations(src)) == [("S028", 6, "E.get", "query:parse_filters")] + + +def test_starred_call_to_a_helper_keeps_its_hand_off() -> None: + src = """\ +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + def get(self, request, *args) -> Response[X]: + return self._use(request.GET, *args) + + def _use(self, params, *args): + return None +""" + assert _summary(_api_violations(src)) == [("S028", 5, "E.get", "query:_use")] + + +def test_helper_shared_by_two_public_methods_is_reported_once() -> None: + src = """\ +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC, "POST": ApiPublishStatus.PUBLIC} + + def get(self, request) -> Response[X]: + return self._shared(request) + + def post(self, request) -> Response[X]: + return self._shared(request) + + def _shared(self, request): + return request.GET.get("k") +""" + violations = _api_violations(src) + assert _summary(violations) == [("S026", 11, "E._shared", "query:k")] + # the first public method in the file that reaches it + assert violations[0].caller == "E.get" + + +def test_helper_read_records_its_caller_and_a_direct_read_does_not() -> None: + src = """\ +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + def get(self, request) -> Response[X]: + request.GET.get("direct") + return build(request) + + +def build(request): + return request.GET.get("inside") +""" + violations = _api_violations(src) + assert _summary(violations) == [ + ("S026", 5, "E.get", "query:direct"), + ("S026", 10, "build", "query:inside"), + ] + assert violations[0].caller is None + # a module-level helper has no class, so its owner comes from the method that reaches it + assert violations[1].caller == "E.get" + + +def test_helper_reached_from_three_methods_is_one_finding() -> None: + src = """\ +class E(Endpoint): + publish_status = { + "GET": ApiPublishStatus.PUBLIC, + "PUT": ApiPublishStatus.PUBLIC, + "DELETE": ApiPublishStatus.PUBLIC, + } + + def get(self, request) -> Response[X]: + return self.filter_things(request) + + def put(self, request) -> Response[X]: + return self.filter_things(request) + + def delete(self, request) -> Response[X]: + return self.filter_things(request) + + def filter_things(self, request): + return request.GET.getlist("id") +""" + assert _summary(_api_violations(src)) == [("S026", 18, "E.filter_things", "query:id")] + + +def test_helper_serializer_stays_on_each_calling_method() -> None: + # S025 depends on the calling method's @extend_schema, so it isn't merged across callers + src = """\ +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC, "PUT": ApiPublishStatus.PUBLIC} + + def get(self, request) -> Response[X]: + return self._validate(request) + + @extend_schema(parameters=[QuerySerializer]) + def put(self, request) -> Response[X]: + return self._validate(request) + + def _validate(self, request): + return QuerySerializer(data=request.GET) +""" + assert _summary(_api_violations(src)) == [("S025", 12, "E.get", "query:QuerySerializer")] + + +def test_helper_with_no_public_caller_is_not_reported() -> None: + src = """\ +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + def get(self, request) -> Response[X]: + return Response() + + def _unused(self, request): + return request.GET.get("k") +""" + assert _api_violations(src) == [] + + +def test_calls_in_decorators_are_not_followed() -> None: + src = """\ +def configure(request): + return request.GET.get("k") + + +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + @some_decorator(configure(request)) + def get(self, request) -> Response[X]: + return Response() +""" + assert _api_violations(src) == [] + + +def test_payload_bound_to_a_renamed_parameter_is_read() -> None: + src = """\ +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + def get(self, request) -> Response[X]: + return self._read(request.GET) + + def _read(self, params): + return params.get("y") +""" + assert _summary(_api_violations(src)) == [("S026", 8, "E._read", "query:y")] + + +def test_request_bound_to_a_renamed_parameter_is_read() -> None: + src = """\ +class E(Endpoint): + publish_status = {"POST": ApiPublishStatus.PUBLIC} + + def post(self, request) -> Response[X]: + return self._read(request) + + def _read(self, req): + return req.data.get("z") +""" + assert _summary(_api_violations(src)) == [("S026", 8, "E._read", "body:z")] + + +def test_same_helper_reads_whatever_each_caller_passes() -> None: + src = """\ +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC, "POST": ApiPublishStatus.PUBLIC} + + def get(self, request) -> Response[X]: + return self._read(request.GET) + + def post(self, request) -> Response[X]: + return self._read(request.data) + + def _read(self, params): + return params.get("k") +""" + assert _summary(_api_violations(src)) == [ + ("S026", 11, "E._read", "query:k"), + ("S026", 11, "E._read", "body:k"), + ] + + +def test_whole_request_is_bound_through_two_levels() -> None: + src = """\ +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + def get(self, request) -> Response[X]: + return self._outer(request) + + def _outer(self, req): + return self._inner(req) + + def _inner(self, r): + return r.GET.get("z") +""" + assert _summary(_api_violations(src)) == [("S026", 11, "E._inner", "query:z")] + + +def test_payload_passed_by_keyword_is_bound() -> None: + src = """\ +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + def get(self, request) -> Response[X]: + return self._read(params=request.GET) + + def _read(self, params): + return params.get("y") +""" + assert _summary(_api_violations(src)) == [("S026", 8, "E._read", "query:y")] + + +def test_staticmethod_helper_has_no_self_to_skip() -> None: + src = """\ +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + def get(self, request) -> Response[X]: + return self._read(request.GET) + + @staticmethod + def _read(params): + return params.get("y") +""" + assert _summary(_api_violations(src)) == [("S026", 9, "E._read", "query:y")] + + +def test_alias_of_a_parameter_is_read() -> None: + src = """\ +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + def get(self, request) -> Response[X]: + self._params(request.GET) + return self._request(request) + + def _params(self, params): + q = params + return q.get("a") + + def _request(self, req): + r = req + return r.GET.get("b") +""" + assert _summary(_api_violations(src)) == [ + ("S026", 10, "E._params", "query:a"), + ("S026", 14, "E._request", "query:b"), + ] + + +def test_parameter_that_is_not_request_input_is_not_reported() -> None: + src = """\ +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + def get(self, request, organization) -> Response[X]: + return self._read(organization) + + def _read(self, org): + return org.get("slug") +""" + assert _api_violations(src) == [] + + +def test_helper_forwarding_its_parameter_to_unknown_code_is_s028() -> None: + src = """\ +class E(Endpoint): + publish_status = {"POST": ApiPublishStatus.PUBLIC} + + def post(self, request) -> Response[X]: + return self._use(request.data) + + def _use(self, params): + return parse_filters(params) +""" + assert _summary(_api_violations(src)) == [("S028", 8, "E._use", "body:parse_filters")] + + +def test_public_method_passing_a_non_request_parameter_keeps_its_s028() -> None: + src = """\ +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + def get(self, request, organization) -> Response[X]: + return serialize(organization, request.GET) +""" + assert _summary(_api_violations(src)) == [("S028", 5, "E.get", "query:serialize")] + + +def test_serializer_built_from_a_helper_parameter_is_checked() -> None: + src = """\ +class E(Endpoint): + publish_status = {"POST": ApiPublishStatus.PUBLIC} + + def post(self, request) -> Response[X]: + return self._validate(request.data) + + def _validate(self, params): + return BodySerializer(data=params) +""" + assert _summary(_api_violations(src)) == [("S025", 8, "E.post", "body:BodySerializer")] + + +def test_payload_that_cannot_be_bound_keeps_its_s028() -> None: + src = """\ +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + def get(self, request) -> Response[X]: + self._extra(None, request.GET) + self._unknown(nope=request.GET) + return self._varargs(request.GET) + + def _extra(self, first): + return None + + def _unknown(self, params): + return None + + def _varargs(self, *args): + return None +""" + assert _summary(_api_violations(src)) == [ + ("S028", 5, "E.get", "query:_extra"), + ("S028", 6, "E.get", "query:_unknown"), + ("S028", 7, "E.get", "query:_varargs"), + ] + + +def _helper_body_summary(body: str) -> list[tuple[ApiRule, int, str, str]]: + """The findings for a GET that passes the query string to `self._h(q)`, with `body` as _h's body""" + src = ( + """\ +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + def get(self, request) -> Response[X]: + return self._h(request.GET) + + def _h(self, q): +""" + + body + ) + return _summary(_api_violations(src)) + + +def test_helper_that_lets_the_payload_leave_keeps_the_s028() -> None: + leaking_bodies = [ + " return q\n", + " return q.copy()\n", + " return q or {}\n", + " return x if c else q\n", + " yield q\n", + " self.q = q\n", + ' d = {}\n d["k"] = q\n return d\n', + ' return {"k": q}\n', + " return [q]\n", + " lst = []\n lst.append(q)\n return lst\n", + " return list(q)\n", + " for k in q:\n print(k)\n", + " q2 = q\n return q2\n", + ' return options.get("k", q)\n', + ] + for body in leaking_bodies: + assert _helper_body_summary(body) == [("S028", 5, "E.get", "query:_h")], body + + +def test_helper_that_only_reads_or_checks_the_payload_is_followed() -> None: + assert _helper_body_summary(' return q.get("x")\n') == [("S026", 8, "E._h", "query:x")] + # isinstance, `in`, `is None`, truthiness and len only look at it + checks = ( + " if not isinstance(q, dict) or q is None or not q or len(q) > 3:\n" + " return None\n" + ' if "x" in q and q:\n' + ' return q["x"]\n' + ) + assert _helper_body_summary(checks) == [("S026", 11, "E._h", "query:x")] + # an alias is checked the same way + assert _helper_body_summary(' q2 = q.copy()\n return q2.get("x")\n') == [ + ("S026", 9, "E._h", "query:x") + ] + + +def test_helper_that_passes_the_payload_to_a_leaking_helper_keeps_that_s028() -> None: + src = """\ +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + def get(self, request) -> Response[X]: + return self._outer(request.GET) + + def _outer(self, q): + return self._inner(q) + + def _inner(self, q): + return q +""" + # _outer only forwards q, so the GET's call is followed; _inner lets it leave, so _outer's call keeps it + assert _summary(_api_violations(src)) == [("S028", 8, "E._outer", "query:_inner")] + + +def test_helpers_passing_the_payload_back_and_forth_are_followed() -> None: + src = """\ +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + def get(self, request) -> Response[X]: + return self._a(request.GET) + + def _a(self, q): + if q.get("again"): + return self._b(q) + return None + + def _b(self, q): + return self._a(q) +""" + assert _summary(_api_violations(src)) == [("S026", 8, "E._a", "query:again")] + + +# Note: when a call could mean more than one function, the helper isn't followed and the call keeps its S028, +# so following helpers never reports LESS than reading each method on its own + + +def test_call_to_a_nested_def_with_a_helpers_name_keeps_the_s028() -> None: + src = """\ +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + def get(self, request) -> Response[X]: + def helper(p): + return send(p) + + return helper(request.GET) + + +def helper(q): + return None +""" + assert _summary(_api_violations(src)) == [("S028", 8, "E.get", "query:helper")] + + +def test_call_to_a_local_import_with_a_helpers_name_keeps_the_s028() -> None: + src = """\ +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + def get(self, request) -> Response[X]: + from elsewhere import helper + + return helper(request.GET) + + +def helper(q): + return q.get("x") +""" + assert _summary(_api_violations(src)) == [("S028", 7, "E.get", "query:helper")] + + +def test_call_to_a_parameter_with_a_helpers_name_keeps_the_s028() -> None: + src = """\ +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + def get(self, request, helper=None) -> Response[X]: + return helper(request.GET) + + +def helper(q): + return q.get("x") +""" + assert _summary(_api_violations(src)) == [("S028", 5, "E.get", "query:helper")] + + +def test_shadowed_name_is_checked_in_the_helper_making_the_call() -> None: + src = """\ +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + def get(self, request) -> Response[X]: + return self._h(request.GET) + + def _h(self, q): + from elsewhere import helper + + return helper(q) + + +def helper(q): + return q.get("x") +""" + assert _summary(_api_violations(src)) == [("S028", 10, "E._h", "query:helper")] + + +def test_helper_defined_twice_keeps_the_s028() -> None: + src = """\ +if FLAG: + def helper(q): + return send(q) +else: + def helper(q): + return None + + +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + def get(self, request) -> Response[X]: + return helper(request.GET) +""" + assert _summary(_api_violations(src)) == [("S028", 13, "E.get", "query:helper")] + + +def test_helper_rebound_at_module_level_keeps_the_s028() -> None: + src = """\ +def helper(q): + return None + + +helper = make_sender() + + +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + def get(self, request) -> Response[X]: + return helper(request.GET) +""" + assert _summary(_api_violations(src)) == [("S028", 12, "E.get", "query:helper")] + + +def test_method_rebound_in_the_class_body_keeps_the_s028() -> None: + src = """\ +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + def get(self, request) -> Response[X]: + return self.h(request.GET) + + def h(self, q): + return None + + h = external_h +""" + assert _summary(_api_violations(src)) == [("S028", 5, "E.get", "query:h")] + + +def test_diamond_class_tree_keeps_the_s028() -> None: + # Python's order is E, B, C, A, so C.h runs; a depth-first walk would pick A.h + src = """\ +class A: + def h(self, q): + return None + + +class B(A): + pass + + +class C(A): + def h(self, q): + return send(q) + + +class E(B, C): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + def get(self, request) -> Response[X]: + return self.h(request.GET) +""" + assert _summary(_api_violations(src)) == [("S028", 19, "E.get", "query:h")] + + +def test_diamond_higher_up_the_class_tree_keeps_the_s028() -> None: + src = """\ +class A: + def h(self, q): + return None + + +class B(A): + pass + + +class C(A): + def h(self, q): + return send(q) + + +class D(B, C): + pass + + +class E(D): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + def get(self, request) -> Response[X]: + return self.h(request.GET) +""" + assert _summary(_api_violations(src)) == [("S028", 23, "E.get", "query:h")] + + +def test_helper_that_reassigns_its_parameter_keeps_the_s028() -> None: + src = """\ +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + def get(self, request) -> Response[X]: + return self._h(request.GET) + + def _h(self, q): + q = {"a": 1} + return q.get("a") +""" + # q no longer holds the query string after it's reassigned, so q.get("a") isn't an S026 + assert _summary(_api_violations(src)) == [("S028", 5, "E.get", "query:_h")] + + +def _post_body_summary(body: str) -> list[tuple[ApiRule, int, str, str]]: + """The findings for a public POST whose body is `body`""" + src = ( + """\ +class E(Endpoint): + publish_status = {"POST": ApiPublishStatus.PUBLIC} + + def post(self, request) -> Response[X]: +""" + + body + ) + return _summary(_api_violations(src)) + + +def test_reads_after_a_local_is_reassigned_are_not_s026() -> None: + body = """\ + data = request.data + serializer = S(data=data) + serializer.is_valid(raise_exception=True) + data = serializer.validated_data + return Response(data.get("name")) +""" + assert _post_body_summary(body) == [("S025", 6, "E.post", "body:S")] + + +def test_reads_before_a_reassignment_are_still_s026() -> None: + body = """\ + data = request.data + first = data.get("a") + data = {} + return Response(data.get("b")) +""" + assert _post_body_summary(body) == [("S026", 6, "E.post", "body:a")] + + +def test_a_value_built_from_request_input_stays_request_input() -> None: + # data["inner"] and clean(data) may still be client input, as master treats them + nested = """\ + data = request.data + data = data["inner"] + return Response(data.get("x")) +""" + assert _post_body_summary(nested) == [ + ("S026", 6, "E.post", "body:inner"), + ("S026", 7, "E.post", "body:x"), + ] + cleaned = """\ + data = request.data + data = clean(data) + return Response(data.get("x")) +""" + assert _post_body_summary(cleaned) == [ + ("S026", 7, "E.post", "body:x"), + ("S028", 6, "E.post", "body:clean"), + ] + + +def test_reassigned_helper_parameter_alias_and_tuple_are_not_s026() -> None: + src = """\ +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + def get(self, request) -> Response[X]: + return self._h(request.GET) + + def _h(self, q): + q2 = q + q2 = {} + first, q = load() + return q.get("a"), q2.get("b") +""" + assert _summary(_api_violations(src)) == [("S028", 5, "E.get", "query:_h")] + + +def test_every_way_of_rebinding_a_local_stops_it_being_request_input() -> None: + rebinds = [ + " x, data = f()\n return data.get('x')\n", + " for data in items:\n data.get('x')\n", + " with open(p) as data:\n data.get('x')\n", + " try:\n pass\n except E as data:\n data.get('x')\n", + " data: dict = {}\n return data.get('x')\n", + ] + for rebind in rebinds: + assert _post_body_summary(" data = request.data\n" + rebind) == [], rebind + + +def test_rebinding_to_another_source_changes_the_source() -> None: + body = """\ + data = request.GET + data = request.data + return Response(data.get("x")) +""" + assert _post_body_summary(body) == [("S026", 7, "E.post", "body:x")] + + +def test_reassignment_inside_a_block_only_applies_inside_it() -> None: + # in the block, data is the validated data + in_block = """\ + data = request.data + if serializer.is_valid(): + data = serializer.validated_data + return Response(data.get("x")) +""" + assert _post_body_summary(in_block) == [] + # after the block it may still be the raw body (when the branch didn't run) + after_block = """\ + data = request.data + if c: + data = serializer.validated_data + return Response(data.get("x")) +""" + assert _post_body_summary(after_block) == [("S026", 8, "E.post", "body:x")] + # the same after a try, and at the top of a loop that reassigns at its end + after_try = """\ + data = request.data + try: + data = parse() + except ValueError: + pass + return Response(data.get("x")) +""" + assert _post_body_summary(after_try) == [("S026", 10, "E.post", "body:x")] + loop_end = """\ + data = request.data + for row in rows: + data.get("x") + data = row +""" + assert _post_body_summary(loop_end) == [("S026", 7, "E.post", "body:x")] + # a nested def's `data = {}` is its own variable + nested_def = """\ + data = request.data + + def build(): + data = {} + return data + + return Response(data.get("x")) +""" + assert _post_body_summary(nested_def) == [("S026", 11, "E.post", "body:x")] + # a source set inside a block is still a source after it + set_in_block = """\ + data = {} + if c: + data = request.data + return Response(data.get("x")) +""" + assert _post_body_summary(set_in_block) == [("S026", 8, "E.post", "body:x")] + + +def test_helper_mixing_a_real_source_and_a_parameter_keeps_both_s028s() -> None: + src = """\ +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + def get(self, request) -> Response[X]: + return self._a(request.data) + + def _a(self, params): + return serialize(params, request.GET) +""" + assert _summary(_api_violations(src)) == [ + ("S028", 8, "E._a", "body:serialize"), + ("S028", 8, "E._a", "query:serialize"), + ] + + +def test_helper_with_two_parameters_forwards_the_payload_one() -> None: + src = """\ +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + def get(self, request, organization) -> Response[X]: + return self._a(organization, request.GET) + + def _a(self, org, params): + return serialize(org, params) +""" + assert _summary(_api_violations(src)) == [("S028", 8, "E._a", "query:serialize")] + + +def test_helper_loop_over_a_declared_serializers_fields_is_silent() -> None: + src = """\ +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + @extend_schema(parameters=[MySerializer]) + def get(self, request) -> Response[X]: + return self._a(request) + + def _a(self, request): + for f in MySerializer().fields: + request.GET.get(f) + return [request.GET.get(name) for name in MySerializer().fields] +""" + assert _api_violations(src) == [] + + +def test_helper_loop_over_an_undeclared_serializers_fields_is_s027() -> None: + src = """\ +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + def get(self, request) -> Response[X]: + return self._a(request) + + def _a(self, request): + for f in MySerializer().fields: + request.GET.get(f) +""" + assert _summary(_api_violations(src)) == [("S027", 9, "E._a", "query:f")] + + +def test_isinstance_check_on_a_payload_is_not_a_hand_off() -> None: + src = """\ +class E(Endpoint): + publish_status = {"POST": ApiPublishStatus.PUBLIC} + + def post(self, request) -> Response[X]: + return self._record(request.data) + + def _record(self, request_data): + if not isinstance(request_data, dict): + return None + return None +""" + assert _api_violations(src) == [] + + +def test_conditional_copy_of_the_query_string_is_an_alias() -> None: + src = """\ +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + def get(self, request) -> Response[X]: + params = request.GET.copy() if request.GET else {} + return params.get("k") +""" + assert _summary(_api_violations(src)) == [("S026", 6, "E.get", "query:k")] + + +def test_date_range_helper_called_inside_a_helper_uses_the_public_schema() -> None: + src = """\ +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + @extend_schema( + parameters=[ + GlobalParams.START, + GlobalParams.END, + GlobalParams.STATS_PERIOD, + GlobalParams.STATS_PERIOD_START, + GlobalParams.STATS_PERIOD_END, + ] + ) + def get(self, request) -> Response[X]: + return self._range(request) + + def _range(self, req): + return get_date_range_from_params(req.GET) +""" + assert _api_violations(src) == [] + + +def test_known_limit_nested_def_reusing_a_parameter_name_reads_the_outer_parameter() -> None: + # A nested def gets no ctx of its own, so its reads count as the helper's. Here inner's + # `params` is the empty dict it's passed, but the read is still bound to _a's `params`, + # which holds the query string. Pinned so it isn't mistaken for a bug. + src = """\ +class E(Endpoint): + publish_status = {"GET": ApiPublishStatus.PUBLIC} + + def get(self, request) -> Response[X]: + return self._a(request.GET) + + def _a(self, params): + def inner(params): + return params.get("x") + + return inner({}) +""" + assert _summary(_api_violations(src)) == [("S026", 9, "E._a", "query:x")] + + def _run_s023(src: str) -> list[str]: return [e for e in _run(src, filename="src/sentry/apidocs/t.py") if "S023" in e] diff --git a/tools/flake8_plugin.py b/tools/flake8_plugin.py index cbf0f7aaf48a..535b6769d6a8 100644 --- a/tools/flake8_plugin.py +++ b/tools/flake8_plugin.py @@ -399,6 +399,39 @@ def _is_iterable_elms_literals( return _is_literal_strings(node) +def _loop_var_name(target: ast.expr) -> str | None: + """The loop variable a for loop or comprehension binds: `for f in ...` or the first of `for f, v in ...`.""" + # check if its a single looping variable + if isinstance(target, ast.Name): + return target.id + # check if its multi looping variables + if isinstance(target, (ast.Tuple, ast.List)) and target.elts: + first_var = target.elts[0] + if isinstance(first_var, ast.Name): + return first_var.id + return None + + +def _fields_serializer(node: ast.expr) -> str | None: + """X for a loop over `X().fields` (or its .keys() / .items()), else None.""" + if ( + isinstance(node, ast.Call) + and not node.args + and not node.keywords + and isinstance(node.func, ast.Attribute) + and node.func.attr in ("keys", "items") + ): + node = node.func.value + if ( + isinstance(node, ast.Attribute) + and node.attr == "fields" + and isinstance(node.value, ast.Call) + and isinstance(node.value.func, ast.Name) + ): + return node.value.func.id + return None + + def _collect_eap_suite_class_names(tree: ast.AST) -> set[str]: """Classes in this module that inherit an EAP events endpoint test base.""" class_bases: dict[str, list[str]] = {} @@ -497,6 +530,7 @@ def extend_schema_kwarg(decorators: list[ast.expr], name: str) -> Generator[ast. "extend", "update", "dumps", + "isinstance", ) ) _COPY_METHODS = frozenset(("copy", "dict")) @@ -509,6 +543,8 @@ def extend_schema_kwarg(decorators: list[ast.expr], name: str) -> Generator[ast. ("START", "END", "STATS_PERIOD", "STATS_PERIOD_START", "STATS_PERIOD_END") ) +_UNRESOLVED: Literal["UNRESOLVED"] = "UNRESOLVED" + def _looks_like_a_class(func: ast.expr) -> bool: """Callee named like a class, which is how a serializer is spelled.""" @@ -516,8 +552,10 @@ def _looks_like_a_class(func: ast.expr) -> bool: def _looks_like_a_class_variable(func: ast.expr) -> bool: - """Callee is a class held in a variable, like serializer_cls which is another way how a serializer is defined.""" - return _name_of(func).endswith("_cls") + """Callee is a serializer class held in a variable or attribute, like serializer_cls, + settings.validator or self.get_serializer, which is another way how a serializer is defined.""" + name = _name_of(func).rsplit(".", 1)[-1].lower() + return name.endswith(("_cls", "serializer", "validator")) def _is_request(node: ast.expr) -> bool: @@ -858,6 +896,41 @@ def _start_and_end(node: ast.expr) -> _StartEnd: ) +@dataclass(frozen=True) +class _FunctionParams: + """A function's parameter names, split by how a call can fill them.""" + + # filled by position in order (posonlyargs + args) + positional: list[str] + # filled by name (args + kwonlyargs) since positional-only params can't be passed by name + keyword: list[str] + + def has(self, name: str) -> bool: + return name in self.positional or name in self.keyword + + +def _function_params(node: ast.FunctionDef | ast.AsyncFunctionDef) -> _FunctionParams: + """A function's parameters. *args and **kwargs are left out, since a payload passed into them + can't be tied to a parameter name.""" + positional: list[str] = [] + keyword: list[str] = [] + for arg in node.args.posonlyargs: + positional.append(arg.arg) + for arg in node.args.args: + positional.append(arg.arg) + keyword.append(arg.arg) + for arg in node.args.kwonlyargs: + keyword.append(arg.arg) + return _FunctionParams(positional, keyword) + + +def _is_staticmethod(node: ast.FunctionDef | ast.AsyncFunctionDef) -> bool: + for decorator in node.decorator_list: + if isinstance(decorator, ast.Name) and decorator.id == "staticmethod": + return True + return False + + @dataclass(frozen=True) class _HelperCall: """One call a function makes to a possible helper""" @@ -876,6 +949,42 @@ class _HelperCall: _SOURCE_PREFIX = {"query string": "query", "request body": "body"} +def _append_once(items: list[Any], item: Any) -> None: + """Append unless already there, since a helper merged twice with different bindings repeats its real reads.""" + if item not in items: + items.append(item) + + +# Sources recorded for a function's own parameters, such as "param:params" or "param:req.GET". +# They never reach a report: the merge resolves them with _resolve_source, or drops them. +_PARAM_PREFIX = "param:" + + +def _resolve_source(source: str, bindings: dict[str, str], for_binding: bool = False) -> str | None: + """Turn a recorded source into a real one, using what this call bound each parameter to. + + Real sources come back unchanged. A placeholder whose parameter didn't receive request + input gives None. A whole request ("request") is only kept `for_binding`, so it can pass + down to the next helper (a read or hand-off of the whole request is silent as before) + """ + if not source.startswith(_PARAM_PREFIX): + return source + param, _, attr = source[len(_PARAM_PREFIX) :].partition(".") + bound = bindings.get(param) + if bound is None: + return None + if attr == "": + if bound == "request" and not for_binding: + return None + return bound + # `.GET` / `.data` read off the parameter which only means something on the whole request + if bound != "request": + return None + if attr == "data": + return "request body" + return "query string" + + @dataclass(frozen=True) class ApiInputViolation: """Structured record of a single S025-S028 API input lint violation. @@ -891,9 +1000,14 @@ class ApiInputViolation: end_line: int end_column: int severity: Literal["error", "warning"] - scope: str # Class.method + # Class.method or function where the read or hand-off is. S025 is the public method instead, + # since it's checked against that method's @extend_schema + scope: str key: str # source and name, for example "query:id" tool: str = "sentry-flake8" + # for a read or hand-off inside a helper, the first public method (in file order) whose request + # reaches it, so a module-level helper, which has no class, can still be given an owner + caller: str | None = None class _InputCtx: @@ -921,17 +1035,50 @@ def __init__( self.computed_reads: list[tuple[_StartEnd, str, str]] = [] # loop variable names currently marked safe (i.e, keys that are known/fixed) self.safe_loop_vars: set[str] = set() + # loop variable -> serializer, for `for f in X().fields` when X isn't declared on this function. + # A helper has no declared names, so whether X is declared is only known at merge time + self.field_loop_vars: dict[str, str] = {} + # (start_end, rendered key, source, serializer) reads keyed by one of those loop variables, + # dropped in the merge when the public method declares the serializer, otherwise S027 + self.field_reads: list[tuple[_StartEnd, str, str, str]] = [] # (start_end, callee, source) the whole dict passed somewhere self.hand_offs: list[tuple[_StartEnd, str, str]] = [] # true only for an HTTP method that publish_status marks PUBLIC on a top-level endpoint class self.is_public = is_public # the class a method is defined in (None for a module-level function) self.class_name = class_name + # this function's parameter names (used to match call arguments to parameters) + self.params = _FunctionParams([], []) + # a @staticmethod has no self to skip when a self.X() call's arguments are matched + self.is_static = False + # local name -> placeholder, for aliases of a parameter such as `q = params` or `q = req.GET` + self.param_locals: dict[str, str] = {} + # names (parameters included) reassigned to something that isn't request input, so reads + # after that point don't count. Reads earlier in the function were already recorded + self.rebound_names: set[str] = set() + # (start_end, serializer, placeholder) for a serializer built from a parameter, + # sorted into query or body validators once the merge knows what the parameter received + self.param_validators: list[tuple[_StartEnd, str, str]] = [] # the calls this function makes to possible helpers self.calls: list[_HelperCall] = [] + # for a merged ctx: the position of each read or hand-off -> the scope of the function it's in + self.found_in: dict[_StartEnd, str] = {} + # this function's def, kept so leaking_params can be worked out only for helpers that need it + self.node: ast.FunctionDef | ast.AsyncFunctionDef | None = None + # parameters whose value can leave this function in a way that isn't recorded as a read or + # hand-off. A call that passes a payload into one of them keeps its S028, since following + # the helper would not show where the payload goes. None until the first call that + # passes a payload into this function asks for it + self.leaking_params: set[str] | None = None + # every name this function binds (parameters, assignments, imports, nested defs...), so a plain + # `helper()` call that could mean a local name isn't followed. None until first needed + self.bound_names: set[str] | None = None def _is(self, node: ast.expr, attrs: frozenset[str], locals_: set[str]) -> bool: node = _unwrap_copy(node) + # `request.GET if c else {}` is a source when either branch is one + if isinstance(node, ast.IfExp): + return self._is(node.body, attrs, locals_) or self._is(node.orelse, attrs, locals_) if isinstance(node, ast.Name): return node.id in locals_ # The attribute has to hang off the request. `serializer.data` and @@ -945,11 +1092,47 @@ def is_body(self, node: ast.expr) -> bool: return self._is(node, frozenset(("data",)), self.body_locals) def source_of(self, node: ast.expr) -> str | None: - """ "query string" / "request body" for an input source, else None.""" + """ "query string" / "request body" for an input source, a placeholder for one of this + function's parameters, else None. Real sources are checked first.""" if self.is_query(node): return "query string" if self.is_body(node): return "request body" + return self.param_source_of(node) + + def forget_request(self, name: str) -> None: + """Stops treating `name` as request input from here on, since it was reassigned. + `name` the local or parameter that was reassigned + Returns nothing, it only updates what this function tracks + """ + self.query_locals.discard(name) + self.body_locals.discard(name) + self.param_locals.pop(name, None) + self.rebound_names.add(name) + + def param_source_of(self, node: ast.expr) -> str | None: + """A placeholder when `node` is one of this function's parameters, or .GET / .data on one. + + `params` -> "param:params", `req.GET` -> "param:req.GET", `req.data` -> "param:req.data". + What the parameter actually holds is only known per call, so the merge resolves it. + """ + node = _unwrap_copy(node) + if isinstance(node, ast.Name): + if node.id in self.param_locals: + return self.param_locals[node.id] + if node.id in self.rebound_names: + return None + if node.id not in ("self", "cls") and self.params.has(node.id): + return f"{_PARAM_PREFIX}{node.id}" + return None + if isinstance(node, ast.Attribute) and (node.attr in _QUERY_ATTRS or node.attr == "data"): + whole = self.param_source_of(node.value) + # .GET / .data only make sense on a whole parameter, not on `req.GET.data` + if whole is None or "." in whole: + return None + if node.attr == "data": + return f"{whole}.data" + return f"{whole}.GET" return None def record_read(self, key: ast.expr, source: str, node: ast.expr) -> None: @@ -957,10 +1140,280 @@ def record_read(self, key: ast.expr, source: str, node: ast.expr) -> None: self.literal_reads.append((_start_and_end(node), key.value, source)) elif isinstance(key, ast.Name) and key.id in self.safe_loop_vars: return + elif isinstance(key, ast.Name) and key.id in self.field_loop_vars: + self.field_reads.append( + (_start_and_end(node), key.id, source, self.field_loop_vars[key.id]) + ) else: self.computed_reads.append((_start_and_end(node), ast.unparse(key), source)) +# Calls that only look at a payload, so passing one doesn't leak it +_PAYLOAD_CHECK_CALLS = frozenset(("isinstance", "len", "bool")) + + +def _parent_map(node: ast.FunctionDef | ast.AsyncFunctionDef) -> dict[ast.AST, ast.AST]: + """Each node in a function's body mapped to the node that contains it. + `node` the function's def + Returns a {child: parent} dict for every node under the body (not the decorators or arguments) + """ + parents: dict[ast.AST, ast.AST] = {} + for statement in node.body: + for parent in ast.walk(statement): + for child in ast.iter_child_nodes(parent): + parents[child] = parent + return parents + + +def _is_test_position(node: ast.AST, parents: dict[ast.AST, ast.AST]) -> bool: + """Whether a value is only checked for truthiness, like `if q:`, `not q` or `if q and xyz`. + `node` the value being checked + `parents` the function's {child: parent} map + Returns True when the value itself can't escape through that expression + """ + parent = parents.get(node) + if isinstance(parent, ast.UnaryOp) and isinstance(parent.op, ast.Not): + return True + while isinstance(parent, ast.BoolOp): + node = parent + parent = parents.get(node) + if isinstance(parent, ast.UnaryOp) and isinstance(parent.op, ast.Not): + return True + if isinstance(parent, (ast.If, ast.While, ast.IfExp, ast.Assert)): + return parent.test is node + return False + + +def _payload_use_is_recorded( + name: ast.Name, parents: dict[ast.AST, ast.AST], aliases: dict[str, str] +) -> bool: + """Whether one use of a parameter (or an alias of it) is a read, hand-off or check the plugin + records or deliberately ignores, so the value can't leave the function unseen. + `name` the use of the parameter + `parents` the function's {child: parent} map + `aliases` local name -> parameter name, the parameters themselves included + Returns False for anything else (returned, yielded, stored, iterated, put in a container, etc.) + """ + node: ast.AST = name + parent = parents.get(node) + # since `q.copy()` / `q.dict()` is still q + while isinstance(parent, ast.Attribute) and parent.attr in _COPY_METHODS: + copy_call = parents.get(parent) + if not (isinstance(copy_call, ast.Call) and copy_call.func is parent): + break + node = copy_call + parent = parents.get(node) + + # q.get("k") / q.getlist("k") / q.pop("k") read a key + if isinstance(parent, ast.Attribute): + call = parents.get(parent) + return parent.attr in _READ_METHODS and isinstance(call, ast.Call) and call.func is parent + # q["k"] + if isinstance(parent, ast.Subscript): + return parent.value is node + # "k" in q, q is None + if isinstance(parent, ast.Compare) and len(parent.ops) == 1: + op = parent.ops[0] + if isinstance(op, (ast.In, ast.NotIn)): + return parent.comparators[0] is node + if isinstance(op, (ast.Is, ast.IsNot)): + other = parent.comparators[0] if parent.left is node else parent.left + return isinstance(other, ast.Constant) and other.value is None + return False + if _is_test_position(node, parents): + return True + # q2 = q, whose own uses are checked too + if isinstance(parent, ast.Assign) and parent.value is node: + for target in parent.targets: + if not (isinstance(target, ast.Name) and target.id in aliases): + return False + return True + + # an argument to a call + keyword = None + if isinstance(parent, ast.keyword): + keyword = parent + parent = parents.get(parent) + if not isinstance(parent, ast.Call) or parent.func is node: + return False + func = parent.func + # options.get("k", q) passes a default which isn't recorded + if isinstance(func, ast.Attribute) and func.attr in _READ_METHODS: + return False + # a serializer's `data =` (the whole call is skipped, so only the data= value counts) + if (_looks_like_a_class(func) or _looks_like_a_class_variable(func)) and any( + kw.arg == "data" for kw in parent.keywords + ): + return keyword is not None and keyword.arg == "data" + callee = _name_of(func).rsplit(".", 1)[-1] + if callee in _PAYLOAD_CHECK_CALLS: + return True + # list(q), sorted(q), lst.append(q), etc copy the payload without recording it + if callee in _CONTAINER_OPS: + return False + # recorded as a hand-off + return True + + +def _payload_leaking_params( + node: ast.FunctionDef | ast.AsyncFunctionDef, ctx: _InputCtx +) -> set[str]: + """The parameters of a function whose value can leave it without being recorded. + `node` the function's def + `ctx` the function's facts, already gathered (its parameters and their aliases) + Returns the names of the leaking parameters + """ + # local name -> the parameter it stands for, the parameters themselves included + aliases: dict[str, str] = {} + for param in ctx.params.positional + ctx.params.keyword: + if param not in ("self", "cls"): + aliases[param] = param + for local, placeholder in ctx.param_locals.items(): + rest = placeholder[len(_PARAM_PREFIX) :] + # only aliases of the whole parameter, not of `req.GET` on a request parameter + if "." not in rest: + aliases[local] = rest + + parents = _parent_map(node) + leaking: set[str] = set() + # how many times each alias is assigned; the assignment that made it is the first + alias_stores: dict[str, int] = {} + for statement in node.body: + for child in ast.walk(statement): + if not (isinstance(child, ast.Name) and child.id in aliases): + continue + if isinstance(child.ctx, ast.Store): + # `q = { xyz }` after `def _h(self, q)` means later uses of q may not be the payload, + # so following the helper can't show where the payload went + if aliases[child.id] == child.id: + leaking.add(child.id) + else: + alias_stores[child.id] = alias_stores.get(child.id, 0) + 1 + if alias_stores[child.id] > 1: + leaking.add(aliases[child.id]) + elif isinstance(child.ctx, ast.Load): + if not _payload_use_is_recorded(child, parents, aliases): + leaking.add(aliases[child.id]) + return leaking + + +def _scope_binding_counts(statements: list[ast.stmt]) -> dict[str, int]: + """How many times each name is bound directly in a module or class body. + `statements` the module's or class's body + Returns {name: count}, counting defs, classes, assignments, imports and every other binding, + including ones under if/try/for, but not the ones inside function or class bodies + """ + counts: dict[str, int] = {} + + def add(name: str) -> None: + counts[name] = counts.get(name, 0) + 1 + + def walk(node: ast.AST) -> None: + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): + add(node.name) + return + if isinstance(node, ast.Lambda): + return + if isinstance(node, (ast.Import, ast.ImportFrom)): + for alias in node.names: + if alias.name != "*": + add(alias.asname or alias.name.split(".")[0]) + return + if isinstance(node, ast.Name) and isinstance(node.ctx, ast.Store): + add(node.id) + if isinstance(node, ast.ExceptHandler) and node.name: + add(node.name) + for child in ast.iter_child_nodes(node): + walk(child) + + for statement in statements: + walk(statement) + return counts + + +def _function_bound_names(node: ast.FunctionDef | ast.AsyncFunctionDef) -> set[str]: + """Every name a function binds anywhere inside it, nested functions included. + `node` the function's def + Returns the names of its parameters, assignment / for / with / except, + imports, nested defs and classes, and global / nonlocal declarations + """ + names: set[str] = set() + for child in ast.walk(node): + if child is node: + continue + if isinstance(child, ast.arg): + names.add(child.arg) + elif isinstance(child, ast.Name) and isinstance(child.ctx, ast.Store): + names.add(child.id) + elif isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): + names.add(child.name) + elif isinstance(child, (ast.Import, ast.ImportFrom)): + for alias in child.names: + names.add(alias.asname or alias.name.split(".")[0]) + elif isinstance(child, ast.ExceptHandler) and child.name: + names.add(child.name) + elif isinstance(child, (ast.Global, ast.Nonlocal)): + for name in child.names: + names.add(name) + return names + + +def _binds_name(ctx: _InputCtx, name: str) -> bool: + """Whether a function binds `name` itself, so a plain `name()` call in it may not be the + module-level function. + `ctx` the function making the call + `name` the called name + Returns True when the function has a parameter, local, import or nested def by that name + """ + if ctx.bound_names is None: + # check if every stored ctx has its def + assert ctx.node is not None + ctx.bound_names = _function_bound_names(ctx.node) + return name in ctx.bound_names + + +def _target_names(target: ast.AST) -> list[str]: + """The plain names an assignment target binds, so `a, (b, *c) = xyz` gives a, b and c + `target` the target being assigned to + Returns the names, or an empty list for attribute and subscript targets + """ + names: list[str] = [] + for node in ast.walk(target): + if isinstance(node, ast.Name) and isinstance(node.ctx, ast.Store): + names.append(node.id) + return names + + +def _touches_source(ctx: _InputCtx, value: ast.expr) -> bool: + """Whether any part of `value` is request input, like `data["inner"]` or `clean(q)`, so the name + it's assigned to may still hold client input and keeps being tracked. + `ctx` the function being visited + `value` the assigned value + Returns True when some part of it is a query string, body or parameter placeholder + """ + for node in ast.walk(value): + if isinstance(node, ast.expr) and ctx.source_of(node) is not None: + return True + return False + + +def _leaks_a_payload(helper_ctx: _InputCtx, bindings: dict[str, str]) -> bool: + """Whether a call hands a payload to a helper parameter it can leave through unseen. + `helper_ctx` the helper being called + `bindings` what each of its parameters received from the call + Returns True when a query string or body lands in one of the helper's leaking parameters + """ + for param, bound in bindings.items(): + if bound not in _SOURCE_PREFIX: + continue + if helper_ctx.leaking_params is None: + assert helper_ctx.node is not None # every stored ctx has its def + helper_ctx.leaking_params = _payload_leaking_params(helper_ctx.node, helper_ctx) + if param in helper_ctx.leaking_params: + return True + return False + + class SentryVisitor(ast.NodeVisitor): def __init__( self, @@ -992,6 +1445,13 @@ def __init__( # Module-level classes, indexed as encountered. A base is always defined # before its subclass at module level, so in-order indexing suffices. self._module_classes: dict[str, ast.ClassDef] = {} + # {name : count of how many times `name` is bound at the module level} filled before the walk + self._module_bindings: dict[str, int] = {} + # { class name : its body's binding counts} filled when a lookup first needs them + self._class_bindings: dict[str, dict[str, int]] = {} + # one frame per block (if / else / loop / try / except body...) being visited in a function: + # name -> what it was tracked as before the block reassigned it, so it can be restored after + self._forget_frames: list[dict[str, tuple[bool, bool, str | None, bool]]] = [] # publish_status of the enclosing module-level class, for S022. self._publish_status: dict[str, str] | None = None # Input rules: per-method accumulation, innermost last. @@ -1007,6 +1467,8 @@ def __init__( self._functions: dict[str, _InputCtx] = {} # every ctx in file order, duplicates included, so a scope defined twice is reported twice self._function_ctxs: list[_InputCtx] = [] + # (rule, scope, key, position) already reported, so a shared helper's line is reported once + self._reported_inputs: set[tuple[ApiRule, str, str, _StartEnd]] = set() def visit_ImportFrom(self, node: ast.ImportFrom) -> None: if node.module == _S030_LEGACY_MODULE and not node.level: @@ -1125,7 +1587,12 @@ def visit_arg(self, node: ast.arg) -> None: def visit_ExceptHandler(self, node: ast.ExceptHandler) -> None: self._except_vars.append(node.name) try: - self.generic_visit(node) + if node.type is not None: + self.visit(node.type) + forget = None + if node.name: + forget = [node.name] + self._visit_block(node.body, forget) finally: self._except_vars.pop() @@ -1152,6 +1619,7 @@ def visit_Try(self, node: ast.Try) -> None: def visit_Module(self, node: ast.Module) -> None: """Visitor for tracking top level constant variables that hold request fields to iterate over.""" self._literal_var_constants = _collect_literal_var_constants(tree=node) + self._module_bindings = _scope_binding_counts(node.body) self.generic_visit(node) self._report_public_inputs() @@ -1166,16 +1634,7 @@ def _new_safe_loop_var( ): return None - loop_var = None - # check if its a single looping variable - if isinstance(target, ast.Name): - loop_var = target.id - - # check if its multi looping variables - elif isinstance(target, (ast.Tuple, ast.List)) and target.elts: - first_var = target.elts[0] - if isinstance(first_var, ast.Name): - loop_var = first_var.id + loop_var = _loop_var_name(target) # an outer loop already marked this loop variable name, but we need to leave it to the outer loop so # the inner loop reusing the name keeps the outer name safe @@ -1184,6 +1643,26 @@ def _new_safe_loop_var( return loop_var + def _new_field_loop_var( + self, target: ast.expr, iter_node: ast.expr, curr_method: _InputCtx + ) -> tuple[str, str] | None: + """(loop variable, serializer) for `for f in X().fields` when X isn't declared on this function. + + Only called when the loop isn't already safe. Whether X is declared is decided in the merge, + against the public method's @extend_schema. + """ + serializer = _fields_serializer(iter_node) + if serializer is None: + return None + loop_var = _loop_var_name(target) + if ( + loop_var is None + or loop_var in curr_method.safe_loop_vars + or loop_var in curr_method.field_loop_vars + ): + return None + return loop_var, serializer + def visit_For(self, node: ast.For) -> None: """Visitor for tracking looping variables declared within a for loop (i.e, `for X in (iterable collection)`, tracking X)""" @@ -1200,14 +1679,23 @@ def visit_For(self, node: ast.For) -> None: if loop_var is not None: curr_method.safe_loop_vars.add(loop_var) - for statement in node.body: - self.visit(statement) + field_var = None + if loop_var is None: + field_var = self._new_field_loop_var(node.target, node.iter, curr_method) + if field_var is not None: + curr_method.field_loop_vars[field_var[0]] = field_var[1] + + loop_forget = None + if not _touches_source(curr_method, node.iter): + loop_forget = _target_names(node.target) + self._visit_block(node.body, forget=loop_forget) if loop_var is not None: curr_method.safe_loop_vars.discard(loop_var) + if field_var is not None: + curr_method.field_loop_vars.pop(field_var[0], None) - for statement in node.orelse: - self.visit(statement) + self._visit_block(node.orelse) def visit_comprehension_types( self, node: ast.ListComp | ast.DictComp | ast.SetComp | ast.GeneratorExp @@ -1221,6 +1709,7 @@ def visit_comprehension_types( curr_method = self._input_stack[-1] valid_vars: list[str] = [] + field_vars: list[str] = [] for comp_node in node.generators: self.visit(comp_node.iter) @@ -1230,6 +1719,11 @@ def visit_comprehension_types( if loop_var is not None: curr_method.safe_loop_vars.add(loop_var) valid_vars.append(loop_var) + else: + field_var = self._new_field_loop_var(comp_node.target, comp_node.iter, curr_method) + if field_var is not None: + curr_method.field_loop_vars[field_var[0]] = field_var[1] + field_vars.append(field_var[0]) for condition in comp_node.ifs: self.visit(condition) @@ -1242,6 +1736,8 @@ def visit_comprehension_types( for var in valid_vars: curr_method.safe_loop_vars.discard(var) + for var in field_vars: + curr_method.field_loop_vars.pop(var, None) visit_ListComp = visit_SetComp = visit_GeneratorExp = visit_DictComp = visit_comprehension_types @@ -1289,13 +1785,18 @@ def _visit_function(self, node: ast.FunctionDef | ast.AsyncFunctionDef) -> None: if ctx is not None: self._input_stack.append(ctx) pushed = True - for statement in node.body: - self.visit(statement) + if pushed: + for statement in node.body: + self.visit(statement) + else: + # a nested def shares the outer function's ctx, so its reassignments end with it + self._visit_block(node.body) finally: self._function_stack.pop() self._function_depth -= 1 if pushed and ctx is not None: self._input_stack.pop() + ctx.node = node self._functions[ctx.scope] = ctx self._function_ctxs.append(ctx) @@ -1311,15 +1812,33 @@ def visit_Assign(self, node: ast.Assign) -> None: for target in node.targets: if isinstance(target, ast.Name): self._s029_seer_pool_names.add(target.id) + # names reassigned to something that isn't request input. The names are forgotten after the value is + # visited, so a read in the value itself (`data = data.get("field")`) still counts + to_forget: list[str] = [] if self._input_stack: ctx = self._input_stack[-1] + # a value built from request input may still be client input, so nothing is forgotten + keeps_input = _touches_source(ctx, node.value) for target in node.targets: if not isinstance(target, ast.Name): + if not keeps_input: + to_forget.extend(_target_names(target)) continue if ctx.is_query(node.value): + self._forget(ctx, target.id) ctx.query_locals.add(target.id) elif ctx.is_body(node.value): + self._forget(ctx, target.id) ctx.body_locals.add(target.id) + else: + # `q = params`, `r = req` or `q = req.GET` alias one of the function's params + placeholder = ctx.param_source_of(node.value) + if placeholder is not None: + self._forget(ctx, target.id) + ctx.param_locals[target.id] = placeholder + ctx.rebound_names.discard(target.id) + elif not keeps_input: + to_forget.append(target.id) if ( _is_tests_path(self.filename) and self._function_depth == 0 @@ -1333,6 +1852,100 @@ def visit_Assign(self, node: ast.Assign) -> None: self._check_S031(node.targets) self._check_S033(node.targets) self.generic_visit(node) + self._forget_names(to_forget) + + def _forget_names(self, names: list[str]) -> None: + """Stops treating each of `names` as request input in the current function, if there is one. + `names` the names that were just reassigned + Returns nothing + """ + if not self._input_stack: + return + for name in names: + self._forget(self._input_stack[-1], name) + + def _forget(self, ctx: _InputCtx, name: str) -> None: + """Forget `name` in `ctx`, first noting what it was in the innermost block, so the block can + put it back when it ends. + `ctx` the function being visited + `name` the name being reassigned + Returns nothing + """ + if self._forget_frames and name not in self._forget_frames[-1]: + self._forget_frames[-1][name] = ( + name in ctx.query_locals, + name in ctx.body_locals, + ctx.param_locals.get(name), + name in ctx.rebound_names, + ) + ctx.forget_request(name) + + def _visit_block(self, statements: list[ast.stmt], forget: list[str] | None = None) -> None: + """Visit one block of statements (an if or else body, a loop body, a try or except body...). + + A reassignment inside the block only applies to the reads after it in the same block. When + the block ends, each name it reassigned is put back to what it was before, on top of anything + the block added, since code after the block may run without the reassignment having happened. + `statements` the block's statements + `forget` names bound when the block starts, like an `except ... as e` name + Returns nothing + """ + if not self._input_stack: + for statement in statements: + self.visit(statement) + return + ctx = self._input_stack[-1] + self._forget_frames.append({}) + try: + if forget: + self._forget_names(forget) + for statement in statements: + self.visit(statement) + finally: + frame = self._forget_frames.pop() + for name, (was_query, was_body, placeholder, was_rebound) in frame.items(): + if was_query: + ctx.query_locals.add(name) + if was_body: + ctx.body_locals.add(name) + if placeholder is not None and name not in ctx.param_locals: + ctx.param_locals[name] = placeholder + if not was_rebound: + ctx.rebound_names.discard(name) + + def generic_visit(self, node: ast.AST) -> None: + """Same as ast.NodeVisitor's, except that inside a function each list of statements is + visited as a block (see _visit_block).""" + if not self._input_stack: + super().generic_visit(node) + return + for _field, value in ast.iter_fields(node): + if isinstance(value, list): + if value and isinstance(value[0], ast.stmt): + self._visit_block(value) + continue + for item in value: + if isinstance(item, ast.AST): + self.visit(item) + elif isinstance(value, ast.AST): + self.visit(value) + + def visit_With(self, node: ast.With) -> None: + self._visit_with(node) + + def visit_AsyncWith(self, node: ast.AsyncWith) -> None: + self._visit_with(node) + + def _visit_with(self, node: ast.With | ast.AsyncWith) -> None: + """`with xyz as data:` rebinds data before the body runs""" + for item in node.items: + self.visit(item) + if item.optional_vars is not None and not ( + self._input_stack and _touches_source(self._input_stack[-1], item.context_expr) + ): + self._forget_names(_target_names(item.optional_vars)) + for statement in node.body: + self.visit(statement) def visit_AugAssign(self, node: ast.AugAssign) -> None: self._check_S031([node.target]) @@ -1343,6 +1956,12 @@ def visit_AnnAssign(self, node: ast.AnnAssign) -> None: self._check_S031([node.target]) self._check_S033([node.target]) self.generic_visit(node) + if ( + node.value is not None + and self._input_stack + and not _touches_source(self._input_stack[-1], node.value) + ): + self._forget_names(_target_names(node.target)) def visit_Delete(self, node: ast.Delete) -> None: self._check_S031(node.targets) @@ -1495,15 +2114,16 @@ def visit_Subscript(self, node: ast.Subscript) -> None: def _new_input_ctx(self, node: ast.FunctionDef | ast.AsyncFunctionDef) -> _InputCtx | None: """Build the accumulator for a top-level function or a method of a top-level class, else None. - Only PUBLIC HTTP methods are reported. The others are stored too, so calls into - them can be followed. The caller pushes it onto _input_stack once the decorators - and arguments have been visited. + Only PUBLIC HTTP methods are reported, every other ctx is a helper whose + "facts" get merged into the public methods that call it. The caller pushes it + onto _input_stack once the decorators and arguments have been visited. """ if self._function_depth != 0: return None if not self._class_stack: ctx = _InputCtx(node.name, set(), set(), is_public=False) + ctx.params = _function_params(node) return ctx if len(self._class_stack) != 1: @@ -1530,6 +2150,8 @@ def _new_input_ctx(self, node: ast.FunctionDef | ast.AsyncFunctionDef) -> _Input is_public=is_public, class_name=self._class_stack[-1], ) + ctx.params = _function_params(node) + ctx.is_static = _is_staticmethod(node) return ctx def _record_input_call(self, node: ast.Call) -> None: @@ -1544,27 +2166,34 @@ def _record_input_call(self, node: ast.Call) -> None: # `options.get("k", request.GET)` passes a default, not the dict. return # Only a serializer's data= is the target shape. This includes uppercase class - # names and dynamic variables ending in `_cls`. `my_func(data=...)` hands the - # dict over exactly as a positional argument would. The `_cls` check is kept - # separate from `_looks_like_a_class` because `_looks_like_a_class` has two - # callers and only this specific check should be expanded to allow `_cls`. + # names, and lowercase names or attributes whose last part ends in `_cls`, + # `serializer` or `validator` (serializer_cls, self.get_serializer, settings.validator). + # `my_func(data=...)` hands the dict over exactly as a positional argument would. + # Known limit: a plain function that happens to end in one of those suffixes, like + # run_validator(data=...), is exempt too. The variable check is kept separate from + # `_looks_like_a_class` because `_looks_like_a_class` has two callers and only this + # specific check should be expanded. if (_looks_like_a_class(func) or _looks_like_a_class_variable(func)) and any( kw.arg == "data" for kw in node.keywords ): return - # The date helpers read a fixed set of keys, so they are only a handoff - # when the endpoint does not document all of them. - if _name_of(func) in _DATE_RANGE_HELPERS and _DATE_RANGE_PARAMS.issubset( - ctx.declared_params - ): + name = _name_of(func).rsplit(".", 1)[-1] + if name in _CONTAINER_OPS: return + # The first real source, as before, plus one hand-off for every parameter placeholder. + # A placeholder only becomes an S028 if the merge finds the parameter received a payload, + # so `serialize(organization, request.GET)` still reports the query string, and in a helper + # `serialize(params, request.GET)` keeps both. The date-range exemption is checked at report time. + recorded_real = False for argument in [*node.args, *(keyword.value for keyword in node.keywords)]: source = ctx.source_of(argument) - if source is not None: - name = _name_of(func).rsplit(".", 1)[-1] - if name not in _CONTAINER_OPS: - ctx.hand_offs.append((_start_and_end(node), name, source)) - return + if source is None: + continue + if source.startswith(_PARAM_PREFIX): + ctx.hand_offs.append((_start_and_end(node), name, source)) + elif not recorded_real: + ctx.hand_offs.append((_start_and_end(node), name, source)) + recorded_real = True def _record_validator(self, node: ast.Call) -> None: """A serializer built from the query string or the request body.""" @@ -1581,6 +2210,10 @@ def _record_validator(self, node: ast.Call) -> None: ctx.query_validators.append((_start_and_end(node), name)) elif ctx.is_body(keyword.value): ctx.body_validators.append((_start_and_end(node), name)) + else: + placeholder = ctx.param_source_of(keyword.value) + if placeholder is not None: + ctx.param_validators.append((_start_and_end(node), name, placeholder)) def _report_input( self, @@ -1591,9 +2224,24 @@ def _report_input( key: str, source: str, rule: ApiRule, + scope: str | None = None, ) -> None: - """Always recorded as a structured violation; fatal only for an enforced group.""" + """Always recorded as a structured violation; fatal only for an enforced group. + + `scope` is where the finding is, when that isn't `ctx` itself (a read inside a helper). A helper + reached from several public methods is reported once, the first time (in file order). + """ enforced = group in ENFORCED + where = scope or ctx.scope + prefixed_key = f"{_SOURCE_PREFIX[source]}:{key}" + # a shared helper's line is one finding, however many public methods reach it + identity = (rule, where, prefixed_key, start_end) + if identity in self._reported_inputs: + return + self._reported_inputs.add(identity) + caller = None + if where != ctx.scope: + caller = ctx.scope self.api_violations.append( ApiInputViolation( rule=rule, @@ -1603,8 +2251,9 @@ def _report_input( end_line=start_end.end_line, end_column=start_end.end_column, severity="error" if enforced else "warning", - scope=ctx.scope, - key=f"{_SOURCE_PREFIX[source]}:{key}", + scope=where, + key=prefixed_key, + caller=caller, ) ) if enforced or os.environ.get(INPUT_LINT_ALL_ENV): @@ -1612,9 +2261,11 @@ def _report_input( def _report_public_inputs(self) -> None: """Report input reads and hand-offs for each public endpoint method after traversing through the complete file""" - for ctx in self._function_ctxs: - if not ctx.is_public: + for http_ctx in self._function_ctxs: + if not http_ctx.is_public: continue + # the HTTP method's own facts plus those of every same-file helper it reaches + ctx = self._merged_http_method_inputs(http_ctx) for start_end, name in ctx.query_validators: if name not in ctx.declared_params: msg = S025_query_msg.format(name) @@ -1627,15 +2278,50 @@ def _report_public_inputs(self) -> None: self._report_input( ctx, start_end, msg, "declared", name, "request body", ApiRule.S025 ) + # S026-S028 are reported where the read or hand-off is, so a helper's line is one finding. + # S025 above stays on the public method, since it depends on that method's @extend_schema for start_end, key, source in ctx.literal_reads: msg = S026_msg.format(repr(key), source) - self._report_input(ctx, start_end, msg, "shaped", key, source, ApiRule.S026) + self._report_input( + ctx, + start_end, + msg, + "shaped", + key, + source, + ApiRule.S026, + ctx.found_in.get(start_end), + ) for start_end, key, source in ctx.computed_reads: msg = S027_msg.format(source, key) - self._report_input(ctx, start_end, msg, "shaped", key, source, ApiRule.S027) + self._report_input( + ctx, + start_end, + msg, + "shaped", + key, + source, + ApiRule.S027, + ctx.found_in.get(start_end), + ) for start_end, callee, source in ctx.hand_offs: + # The date helpers read a fixed set of keys, so they are only a hand-off + # when the endpoint does not document all of them. + if callee in _DATE_RANGE_HELPERS and _DATE_RANGE_PARAMS.issubset( + ctx.declared_params + ): + continue msg = S028_msg.format(source, callee) - self._report_input(ctx, start_end, msg, "shaped", callee, source, ApiRule.S028) + self._report_input( + ctx, + start_end, + msg, + "shaped", + callee, + source, + ApiRule.S028, + ctx.found_in.get(start_end), + ) def _request_payload_input(self, ctx: _InputCtx, argument: ast.expr) -> str | None: """Returns what request input the argument carries: "request", "query string", "request body", or None""" @@ -1715,6 +2401,263 @@ def _record_helper(self, node: ast.Call) -> None: _HelperCall(kind, name, request_params, starred, _start_and_end(node)) ) + def _helper_call_ctx_lookup( + self, helper_call: _HelperCall, endpoint_class: str, helper_defined_class: str | None + ) -> _InputCtx | Literal["UNRESOLVED"] | None: + """Return the ctx of the same-file function that `helper_call` runs. + + `endpoint_class` is the endpoint class the helper was called in. + `helper_defined_class` is the class that defines the helper function making the `helper_call`. + + Returns None if the name isn't defined in this file, and _UNRESOLVED if the lookup reaches a class from another file before finding the method. + """ + if helper_call.kind == Kind.NORMAL_FUNCTION: + # a name bound more than once (two defs, or a def then `helper = xyz`) could be either + if self._module_bindings.get(helper_call.name, 0) != 1: + return None + return self._functions.get(helper_call.name) + + order = self._handle_inheritance_method_lookup(public_class=endpoint_class) + if helper_call.kind == Kind.PARENT_CLASS_METHOD: + if helper_defined_class not in order: + return None + start = order.index(helper_defined_class) + # which class super() continues with depends on the whole tree below it + for cls in order[: start + 1]: + if self._is_ambiguous_class(cls): + return _UNRESOLVED + order = order[start + 1 :] + + for cls in order: + if cls not in self._module_classes or self._module_bindings.get(cls, 0) != 1: + return _UNRESOLVED + bound = self._class_binding_count(cls, helper_call.name) + if bound == 1: + ctx = self._functions.get(f"{cls}.{helper_call.name}") + # bound once, but not by a def (i.e, `h = external_h`) + if ctx is None: + return _UNRESOLVED + return ctx + if bound > 1: + return _UNRESOLVED + # past this class the depth-first order may not be the real method order so we need to check if it's an ambiguous class + if self._is_ambiguous_class(cls): + return _UNRESOLVED + + return None + + def _class_binding_count(self, cls: str, name: str) -> int: + """How many times `name` is bound directly in the body of the module-level class `cls`.""" + if cls not in self._class_bindings: + self._class_bindings[cls] = _scope_binding_counts(self._module_classes[cls].body) + return self._class_bindings[cls].get(name, 0) + + def _is_ambiguous_class(self, cls: str) -> bool: + """Whether `cls` has more than one base from this file, so the classes after it in the + depth-first walk may not be in Python's real method order (a diamond, for example).""" + class_node = self._module_classes.get(cls) + if class_node is None: + return False + same_file = 0 + for base in class_node.bases: + if _name_of(base) in self._module_classes: + same_file += 1 + return same_file > 1 + + def _handle_inheritance_method_lookup(self, public_class: str) -> list[str]: + """Returns class names in depth-first method lookup order starting from `public_class` all the way up to its parents to deal with inheritance. + `public_class` the starting endpoint class name whose inheritance chain is being searched + + Class Grandfather() + + Class Father(Grandfather) + + Class Child(Father) + + _handle_inheritance_method_lookup("Child") -> ["Child", "Father", "Grandfather"] + + A super() call starts after the class that made it, which _helper_call_ctx_lookup handles. + """ + order: list[str] = [] + self._walk_class_bases(name=public_class, visited=set(), order=order) + return order + + def _walk_class_bases(self, name: str, visited: set[str], order: list[str]) -> None: + """Appends `name` and each of its bases depth first, left to right, and each class once. + `name` the name of the current class being visited in the inheritance tree + `visited` set of class names already seen during the walk + `order` list that collects class names in depth first, left to right lookup order as the recursion walks up the parents + """ + if name in visited: + return + visited.add(name) + order.append(name) + + class_node = self._module_classes.get(name) + if class_node is None: + return + + for base in class_node.bases: + base_name = _name_of(base) + if base_name != "object": + self._walk_class_bases(base_name, visited, order) + + def _merged_http_method_inputs(self, http_ctx: _InputCtx) -> _InputCtx: + """Return a new ctx with `http_ctx`'s own facts plus those of every same-file helper it reaches. + Stored ctxs are never changed, so a helper shared by two public methods is merged into each separately. + _report_public_inputs() runs S025-S028 on the result, so reads inside helpers are reported under the HTTP method's `http_ctx`. + """ + assert http_ctx.class_name is not None # to satisfy mypy + + # create the ctx based off of the http method (its our ctx that will hold everything the http method ctx holds + the helpers ctx holds) + new_ctx = _InputCtx( + scope=http_ctx.scope, + declared_params=http_ctx.declared_params, + declared_body=http_ctx.declared_body, + is_public=True, + class_name=http_ctx.class_name, + ) + + # the HTTP method's own reads are already real sources, so it starts with no bindings + self._merge_ctx_into( + new_ctx=new_ctx, + ctx=http_ctx, + seen={(http_ctx.scope, frozenset())}, + bindings={}, + ) + + return new_ctx + + def _bind_call( + self, call: _HelperCall, target: _InputCtx, caller_bindings: dict[str, str] + ) -> tuple[dict[str, str], bool]: + """What each of `target`'s parameters received from `call`, in real terms + ("request" / "query string" / "request body"), and whether every payload found a named parameter. + + A payload that lands past the positional parameters, on a keyword the def doesn't have, or in + *args / **kwargs can't be tied to a name, so the call isn't fully bound and its S028 stays. + """ + positional = target.params.positional + # self.X() and super().X() fill self (or cls) themselves, unless the target is a @staticmethod + if call.kind != Kind.NORMAL_FUNCTION and not target.is_static: + positional = positional[1:] + + bindings: dict[str, str] = {} + all_bound = True + for where, what in call.passes: + # what the caller passed, resolved with the caller's own bindings + real = _resolve_source(what, caller_bindings, for_binding=True) + # not request input for this caller, i.e organization + if real is None: + continue + if isinstance(where, int): + if where >= len(positional): + all_bound = False + continue + bindings[positional[where]] = real + else: + if where not in target.params.keyword: + all_bound = False + continue + bindings[where] = real + return bindings, all_bound + + def _merge_ctx_into( + self, + new_ctx: _InputCtx, + ctx: _InputCtx, + seen: set[tuple[str, frozenset[tuple[str, str]]]], + bindings: dict[str, str], + ) -> None: + """Copies `ctx`'s facts (i.e. `literal_reads`, `hand_offs`, `computed_reads`) into `new_ctx`, then do the same for each same-file helper + `ctx` calls. + `seen` holds (scope, bindings) already copied into `new_ctx` for this HTTP method, starting with its own, so the same + helper called with different inputs is still followed. + `bindings` is what each of `ctx`'s parameters received, used to turn its placeholders into real sources. + """ + assert new_ctx.class_name is not None # satisfying mypy + + # positions of calls whose helper was followed with every payload so their S028 hand-off is dropped + followed: set[_StartEnd] = set() + + # (helper ctx, what its parameters received) to merge next + to_visit: list[tuple[_InputCtx, dict[str, str]]] = [] + + # loop over all the helper calls this function makes + for call in ctx.calls: + if call.starred: + continue + # `helper()` in a function with its own `helper` (a parameter, import, nested def, etc) but it may + # not be the module-level one + if call.kind == Kind.NORMAL_FUNCTION and _binds_name(ctx, call.name): + continue + + # gets the ctx of the helper + helper_ctx = self._helper_call_ctx_lookup( + helper_call=call, + endpoint_class=new_ctx.class_name, + helper_defined_class=ctx.class_name, + ) + + # skip it unless it's a helper in the same file (None = not in this file, _UNRESOLVED = reached another file) + if not isinstance(helper_ctx, _InputCtx): + continue + + helper_bindings, all_bound = self._bind_call(call, helper_ctx, bindings) + # a payload that couldn't be tied to a parameter, or that can leave the helper unseen + # (i.e `return params`), keeps its S028 + if all_bound and not _leaks_a_payload(helper_ctx, helper_bindings): + followed.add(call.at) + + to_visit.append((helper_ctx, helper_bindings)) + + # copy the facts over, turning placeholders into real sources and dropping the ones that don't resolve + for start_end, name in ctx.query_validators: + _append_once(new_ctx.query_validators, (start_end, name)) + for start_end, name in ctx.body_validators: + _append_once(new_ctx.body_validators, (start_end, name)) + for start_end, name, placeholder in ctx.param_validators: + real = _resolve_source(placeholder, bindings) + if real == "query string": + _append_once(new_ctx.query_validators, (start_end, name)) + elif real == "request body": + _append_once(new_ctx.body_validators, (start_end, name)) + for start_end, key, source in ctx.literal_reads: + real = _resolve_source(source, bindings) + if real is not None: + _append_once(new_ctx.literal_reads, (start_end, key, real)) + new_ctx.found_in[start_end] = ctx.scope + for start_end, key, source in ctx.computed_reads: + real = _resolve_source(source, bindings) + if real is not None: + _append_once(new_ctx.computed_reads, (start_end, key, real)) + new_ctx.found_in[start_end] = ctx.scope + # a loop over X().fields reads a fixed set of keys when the public method declares X + for start_end, key, source, serializer in ctx.field_reads: + if serializer in new_ctx.declared_params or serializer in new_ctx.declared_body: + continue + real = _resolve_source(source, bindings) + if real is not None: + _append_once(new_ctx.computed_reads, (start_end, key, real)) + new_ctx.found_in[start_end] = ctx.scope + for start_end, callee, source in ctx.hand_offs: + if start_end in followed: + continue + real = _resolve_source(source, bindings) + if real is not None: + _append_once(new_ctx.hand_offs, (start_end, callee, real)) + new_ctx.found_in[start_end] = ctx.scope + + for helper_ctx, helper_bindings in to_visit: + seen_key = (helper_ctx.scope, frozenset(helper_bindings.items())) + if seen_key in seen: + continue + + seen.add(seen_key) + self._merge_ctx_into( + new_ctx=new_ctx, ctx=helper_ctx, seen=seen, bindings=helper_bindings + ) + def visit_Call(self, node: ast.Call) -> None: self._s024_visit_call(node) if legacy_name := self._s030_call_name(node.func):