feat(apidocs): Lint request parameters read without a serializer

setup.cfg

@@ -70,9 +70,9 @@ per-file-ignores =
     tools/*: S
     # testing the options manager itself
     src/sentry/testutils/helpers/options.py, tests/sentry/options/test_manager.py: S011
-    # S021-S025 lint the shipped API surface and its lint infrastructure; test
+    # S021-S028 lint the shipped API surface and its lint infrastructure; test
     # modules deliberately contain the shapes they exercise
-    tests/*: S021, S022, S023, S024, S025
+    tests/*: S021, S022, S023, S024, S025, S026, S027, S028
 
 [flake8:local-plugins]
 paths = .

tests/tools/test_flake8_plugin.py

@@ -1224,7 +1224,7 @@ def _run_input(src: str, enforced: frozenset[str] = frozenset({"declared"})) ->
         return [
             e
             for e in _run(src, filename="src/sentry/api/endpoints/t.py")
-            if "S025" in e or "S026" in e or "S027" in e
+            if any(code in e for code in ("S025", "S026", "S027", "S028"))
         ]
     finally:
         plugin.ENFORCED = original
@@ -1388,3 +1388,225 @@ def post(self, request) -> Response[X]:
         UploadSerializer(data=request.data)
 """
     assert _run_input(src) == []
+
+
+SHAPED = frozenset({"shaped"})
+
+
+def test_S026_raw_query_read_is_reported() -> None:
+    src = """\
+class E(Endpoint):
+    publish_status = {"GET": ApiPublishStatus.PUBLIC}
+
+    def get(self, request) -> Response[X]:
+        return request.GET.get("truncate")
+"""
+    errors = _run_input(src, SHAPED)
+    assert errors == [
+        "t.py:5:15: S026 'truncate' is read straight off the query string, so the schema has "
+        "nothing to document and the value is an unchecked string. Read it through a "
+        "serializer declared in @extend_schema."
+    ]
+
+
+def test_S026_raw_body_read_is_reported() -> None:
+    src = """\
+class E(Endpoint):
+    publish_status = {"POST": ApiPublishStatus.PUBLIC}
+
+    def post(self, request) -> Response[X]:
+        return request.data["origin"]
+"""
+    errors = _run_input(src, SHAPED)
+    assert len(errors) == 1
+    assert "'origin' is read straight off the request body" in errors[0]
+
+
+def test_S026_subscript_and_getlist_are_both_reads() -> None:
+    src = """\
+class E(Endpoint):
+    publish_status = {"GET": ApiPublishStatus.PUBLIC}
+
+    def get(self, request) -> Response[X]:
+        a = request.GET["one"]
+        b = request.GET.getlist("two")
+        return a, b
+"""
+    assert len(_run_input(src, SHAPED)) == 2
+
+
+def test_S026_read_through_validated_data_is_accepted() -> None:
+    src = """\
+class E(Endpoint):
+    publish_status = {"GET": ApiPublishStatus.PUBLIC}
+
+    @extend_schema(parameters=[QuerySerializer])
+    def get(self, request) -> Response[X]:
+        serializer = QuerySerializer(data=request.GET)
+        return serializer.validated_data["truncate"]
+"""
+    assert _run_input(src, SHAPED) == []
+
+
+def test_S026_private_method_is_skipped() -> None:
+    src = """\
+class E(Endpoint):
+    publish_status = {"GET": ApiPublishStatus.PRIVATE}
+
+    def get(self, request) -> Response[X]:
+        return request.GET.get("truncate")
+"""
+    assert _run_input(src, SHAPED) == []
+
+
+def test_S027_computed_key_is_reported() -> None:
+    src = """\
+class E(Endpoint):
+    publish_status = {"GET": ApiPublishStatus.PUBLIC}
+
+    def get(self, request) -> Response[X]:
+        return request.GET.get(some_name)
+"""
+    errors = _run_input(src, SHAPED)
+    assert len(errors) == 1
+    assert errors[0].startswith(
+        "t.py:5:15: S027 the query string is read with the computed key some_name"
+    )
+
+
+def test_S028_hand_off_is_reported() -> None:
+    src = """\
+class E(Endpoint):
+    publish_status = {"GET": ApiPublishStatus.PUBLIC}
+
+    def get(self, request) -> Response[X]:
+        return installation.get_link_issue_config(params=request.GET)
+"""
+    errors = _run_input(src, SHAPED)
+    assert len(errors) == 1
+    assert "handed to get_link_issue_config" in errors[0]
+
+
+def test_S028_container_operations_are_not_hand_offs() -> None:
+    src = """\
+class E(Endpoint):
+    publish_status = {"GET": ApiPublishStatus.PUBLIC}
+
+    def get(self, request) -> Response[X]:
+        return len(request.GET), sorted(request.GET)
+"""
+    assert _run_input(src, SHAPED) == []
+
+
+def test_S028_building_a_serializer_is_not_a_hand_off() -> None:
+    src = """\
+class E(Endpoint):
+    publish_status = {"GET": ApiPublishStatus.PUBLIC}
+
+    @extend_schema(parameters=[QuerySerializer])
+    def get(self, request) -> Response[X]:
+        return QuerySerializer(data=request.GET)
+"""
+    assert _run_input(src, SHAPED) == []
+
+
+def test_shaped_rule_records_instead_of_gating_when_unenforced() -> None:
+    src = """\
+class E(Endpoint):
+    publish_status = {"GET": ApiPublishStatus.PUBLIC}
+
+    def get(self, request) -> Response[X]:
+        return request.GET.get("truncate")
+"""
+    assert _run_input(src, frozenset()) == []
+
+
+def test_rules_are_enabled_independently() -> None:
+    src = """\
+class E(Endpoint):
+    publish_status = {"GET": ApiPublishStatus.PUBLIC}
+
+    def get(self, request) -> Response[X]:
+        QuerySerializer(data=request.GET)
+        return request.GET.get("truncate")
+"""
+    declared_only = _run_input(src, frozenset({"declared"}))
+    assert len(declared_only) == 1 and "S025" in declared_only[0]
+    shaped_only = _run_input(src, SHAPED)
+    assert len(shaped_only) == 1 and "S026" in shaped_only[0]
+
+
+def test_S026_serializer_and_response_data_are_not_request_body() -> None:
+    src = """\
+class E(Endpoint):
+    publish_status = {"POST": ApiPublishStatus.PUBLIC}
+
+    def post(self, request) -> Response[X]:
+        serializer.data["title"]
+        response.data["title"]
+        return serializer.data.get("slug")
+"""
+    assert _run_input(src, SHAPED) == []
+
+
+def test_S026_request_via_self_is_still_a_read() -> None:
+    src = """\
+class E(Endpoint):
+    publish_status = {"GET": ApiPublishStatus.PUBLIC}
+
+    def get(self, request) -> Response[X]:
+        return self.request.GET.get("truncate")
+"""
+    assert len(_run_input(src, SHAPED)) == 1
+
+
+def test_S026_subscript_write_is_not_a_read() -> None:
+    src = """\
+class E(Endpoint):
+    publish_status = {"POST": ApiPublishStatus.PUBLIC}
+
+    def post(self, request) -> Response[X]:
+        request.data["title"] = "default"
+        del request.data["scratch"]
+        return request.data["title"]
+"""
+    errors = _run_input(src, SHAPED)
+    assert len(errors) == 1
+    assert "t.py:7:" in errors[0]
+
+
+def test_S028_plain_function_taking_data_is_still_a_hand_off() -> None:
+    src = """\
+class E(Endpoint):
+    publish_status = {"GET": ApiPublishStatus.PUBLIC}
+
+    def get(self, request) -> Response[X]:
+        return my_func(data=request.GET)
+"""
+    errors = _run_input(src, SHAPED)
+    assert len(errors) == 1
+    assert "handed to my_func" in errors[0]
+
+
+def test_S028_method_taking_data_is_still_a_hand_off() -> None:
+    src = """\
+class E(Endpoint):
+    publish_status = {"POST": ApiPublishStatus.PUBLIC}
+
+    def post(self, request) -> Response[X]:
+        return installation.build(data=request.data)
+"""
+    errors = _run_input(src, SHAPED)
+    assert len(errors) == 1
+    assert "handed to build" in errors[0]
+
+
+def test_S028_request_data_as_a_lookup_default_is_not_a_hand_off() -> None:
+    src = """\
+class E(Endpoint):
+    publish_status = {"GET": ApiPublishStatus.PUBLIC}
+
+    def get(self, request) -> Response[X]:
+        return options.get("key", request.GET)
+"""
+    assert _run_input(src, SHAPED) == []