diff --git a/setup.cfg b/setup.cfg
index 56c4db0e942b..6d846bd497a1 100644
--- a/setup.cfg
+++ b/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 = .
diff --git a/tests/tools/test_flake8_plugin.py b/tests/tools/test_flake8_plugin.py
index 56dbb4d4578b..e65069c56d42 100644
--- a/tests/tools/test_flake8_plugin.py
+++ b/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) == []
diff --git a/tools/flake8_plugin.py b/tools/flake8_plugin.py
index c65c2d663380..db108521a1ff 100644
--- a/tools/flake8_plugin.py
+++ b/tools/flake8_plugin.py
@@ -168,6 +168,19 @@
     "@extend_schema(parameters=...), so the schema does not document what this "
     "endpoint accepts. Add it to parameters=."
 )
+S026_msg = (
+    "S026 {} is read straight off the {}, so the schema has nothing to document and "
+    "the value is an unchecked string. Read it through a serializer declared in "
+    "@extend_schema."
+)
+S027_msg = (
+    "S027 the {} is read with the computed key {}, so no schema can document it. "
+    "Read a literal key, or declare this endpoint's input as an exception."
+)
+S028_msg = (
+    "S028 the whole {} is handed to {}, so what this endpoint accepts cannot be "
+    "determined. Read the values here, or declare this endpoint's input as an exception."
+)
 S025_body_msg = (
     "S025 {} validates the request body but is not declared in "
     "@extend_schema(request=...), so the schema does not document what this "
@@ -309,9 +322,42 @@ def extend_schema_kwarg(decorators: list[ast.expr], name: str) -> Generator[ast.
 
 
 _QUERY_ATTRS = frozenset(("GET", "query_params"))
+_READ_METHODS = frozenset(("get", "getlist", "pop"))
+# Counting or iterating the dict does not read a parameter out of it, so these
+# are not hand-offs. Anything else receiving the whole dict might read anything.
+_CONTAINER_OPS = frozenset(
+    (
+        "len",
+        "list",
+        "set",
+        "tuple",
+        "sorted",
+        "dict",
+        "bool",
+        "any",
+        "all",
+        "iter",
+        "append",
+        "extend",
+        "update",
+        "dumps",
+    )
+)
 _COPY_METHODS = frozenset(("copy", "dict"))
 
 
+def _looks_like_a_class(func: ast.expr) -> bool:
+    """Callee named like a class, which is how a serializer is spelled."""
+    return _name_of(func).rsplit(".", 1)[-1][:1].isupper()
+
+
+def _is_request(node: ast.expr) -> bool:
+    """The handler's request argument, as `request` or `self.request`."""
+    if isinstance(node, ast.Name):
+        return node.id == "request"
+    return isinstance(node, ast.Attribute) and node.attr == "request"
+
+
 def _unwrap_copy(node: ast.expr) -> ast.expr:
     """Strip `.copy()` / `.dict()` so `request.GET.copy()` still reads as the source."""
     while (
@@ -582,12 +628,20 @@ def __init__(self, declared_params: set[str], declared_body: set[str]) -> None:
         # (line, col, serializer) for each serializer built from that source
         self.query_validators: list[tuple[int, int, str]] = []
         self.body_validators: list[tuple[int, int, str]] = []
+        # (line, col, key, source) reads with a literal key
+        self.literal_reads: list[tuple[int, int, str, str]] = []
+        # (line, col, rendered key, source) reads whose key is computed
+        self.computed_reads: list[tuple[int, int, str, str]] = []
+        # (line, col, callee, source) the whole dict passed somewhere
+        self.hand_offs: list[tuple[int, int, str, str]] = []
 
     def _is(self, node: ast.expr, attrs: frozenset[str], locals_: set[str]) -> bool:
         node = _unwrap_copy(node)
         if isinstance(node, ast.Name):
             return node.id in locals_
-        return isinstance(node, ast.Attribute) and node.attr in attrs
+        # The attribute has to hang off the request. `serializer.data` and
+        # `response.data` are outputs, not parameters a client sent.
+        return isinstance(node, ast.Attribute) and node.attr in attrs and _is_request(node.value)
 
     def is_query(self, node: ast.expr) -> bool:
         return self._is(node, _QUERY_ATTRS, self.query_locals)
@@ -595,6 +649,20 @@ def is_query(self, node: ast.expr) -> bool:
     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."""
+        if self.is_query(node):
+            return "query string"
+        if self.is_body(node):
+            return "request body"
+        return None
+
+    def record_read(self, key: ast.expr, source: str, line: int, col: int) -> None:
+        if isinstance(key, ast.Constant) and isinstance(key.value, str):
+            self.literal_reads.append((line, col, key.value, source))
+        else:
+            self.computed_reads.append((line, col, ast.unparse(key), source))
+
 
 class SentryVisitor(ast.NodeVisitor):
     def __init__(
@@ -923,6 +991,16 @@ def _s024_visit_call(self, node: ast.Call) -> None:
             if self._s024_parses is None:
                 self._s024_parses = (node.lineno, node.col_offset)
 
+    def visit_Subscript(self, node: ast.Subscript) -> None:
+        # Load only: `request.data["title"] = ...` writes a value, it does not
+        # read a parameter the client sent.
+        if self._input_stack and isinstance(node.ctx, ast.Load):
+            ctx = self._input_stack[-1]
+            source = ctx.source_of(node.value)
+            if source is not None:
+                ctx.record_read(node.slice, source, node.lineno, node.col_offset)
+        self.generic_visit(node)
+
     def _enter_input(self, node: ast.FunctionDef | ast.AsyncFunctionDef) -> bool:
         """Push an accumulator for a PUBLIC HTTP method on an endpoint class."""
         if len(self._class_stack) != 1 or self._function_depth != 0:
@@ -940,17 +1018,40 @@ def _enter_input(self, node: ast.FunctionDef | ast.AsyncFunctionDef) -> bool:
         )
         return True
 
+    def _record_input_call(self, node: ast.Call) -> None:
+        """A `.get()` read, or the whole dict handed to something else."""
+        ctx = self._input_stack[-1]
+        func = node.func
+        if isinstance(func, ast.Attribute) and func.attr in _READ_METHODS:
+            source = ctx.source_of(func.value)
+            if source is not None and node.args:
+                ctx.record_read(node.args[0], source, node.lineno, node.col_offset)
+            # On anything but the request this is an ordinary lookup, and
+            # `options.get("k", request.GET)` passes a default, not the dict.
+            return
+        # Only a serializer's data= is the target shape. `my_func(data=...)`
+        # hands the dict over exactly as a positional argument would.
+        if _looks_like_a_class(func) and any(kw.arg == "data" for kw in node.keywords):
+            return
+        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((node.lineno, node.col_offset, name, source))
+                return
+
     def _record_validator(self, node: ast.Call) -> None:
         """A serializer built from the query string or the request body."""
         ctx = self._input_stack[-1]
         for keyword in node.keywords:
             if keyword.arg != "data":
                 continue
-            name = _name_of(node.func).rsplit(".", 1)[-1]
             # A class, by convention. Skips plain calls taking data=, and
             # runtime-chosen classes the schema could not name either.
-            if not name[:1].isupper():
+            if not _looks_like_a_class(node.func):
                 continue
+            name = _name_of(node.func).rsplit(".", 1)[-1]
             if ctx.is_query(keyword.value):
                 ctx.query_validators.append((node.lineno, node.col_offset, name))
             elif ctx.is_body(keyword.value):
@@ -969,11 +1070,18 @@ def _exit_input(self) -> None:
         for line, col, name in ctx.body_validators:
             if name not in ctx.declared_body:
                 self._report_input(line, col, S025_body_msg.format(name), "declared")
+        for line, col, key, source in ctx.literal_reads:
+            self._report_input(line, col, S026_msg.format(repr(key), source), "shaped")
+        for line, col, key, source in ctx.computed_reads:
+            self._report_input(line, col, S027_msg.format(source, key), "shaped")
+        for line, col, callee, source in ctx.hand_offs:
+            self._report_input(line, col, S028_msg.format(source, callee), "shaped")
 
     def visit_Call(self, node: ast.Call) -> None:
         self._s024_visit_call(node)
         if self._input_stack:
             self._record_validator(node)
+            self._record_input_call(node)
         if _is_tests_path(self.filename):
             if (
                 isinstance(node.func, ast.Name)