[v3-1-test] Fix recursion depth error in _redact_exception_with_context (#61776)

* Fix recursion depth error in _redact_exception_with_context

Fix recursion depth check in _redact_exception_with_context_or_cause.
There are some obscure cases where exception might point to itself
in cause/context - this PR protects against it.

Changed name to include cause as well.

Initially implemented as a fix to v2-11-test in #61254 - enhanced
with the case of removal of too-deep exceptions rather than
not-redacting it (and replacing it with sentinel exception explaining
that reminder of the stack trace has been removed.

Co-authored-by: Anton Nitochkin <nitochkin@google.com>

* Apply suggestions from code review

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

---------
(cherry picked from commit 1533ecf)

Co-authored-by: Jarek Potiuk <jarek@potiuk.com>
Co-authored-by: Anton Nitochkin <nitochkin@google.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Author: 3 people

shared/secrets_masker/src/airflow_shared/secrets_masker/secrets_masker.py

@@ -263,15 +263,38 @@ def _record_attrs_to_ignore(self) -> Iterable[str]:
         )
         return frozenset(record.__dict__).difference({"msg", "args"})
 
-    def _redact_exception_with_context(self, exception):
+    def _redact_exception_with_context_or_cause(self, exception, visited=None):
         # Exception class may not be modifiable (e.g. declared by an
         # extension module such as JDBC).
         with contextlib.suppress(AttributeError):
-            exception.args = (self.redact(v) for v in exception.args)
-        if exception.__context__:
-            self._redact_exception_with_context(exception.__context__)
-        if exception.__cause__ and exception.__cause__ is not exception.__context__:
-            self._redact_exception_with_context(exception.__cause__)
+            if visited is None:
+                visited = set()
+
+            if id(exception) in visited:
+                # already visited - it was redacted earlier
+                return exception
+
+            # Check depth before adding to visited to ensure we skip exceptions beyond the limit
+            if len(visited) >= self.MAX_RECURSION_DEPTH:
+                return RuntimeError(
+                    f"Stack trace redaction hit recursion limit of {self.MAX_RECURSION_DEPTH} "
+                    f"when processing exception of type {type(exception).__name__}. "
+                    f"The remaining exceptions will be skipped to avoid "
+                    f"infinite recursion and protect against revealing sensitive information."
+                )
+
+            visited.add(id(exception))
+
+            exception.args = tuple(self.redact(v) for v in exception.args)
+            if exception.__context__:
+                exception.__context__ = self._redact_exception_with_context_or_cause(
+                    exception.__context__, visited
+                )
+            if exception.__cause__ and exception.__cause__ is not exception.__context__:
+                exception.__cause__ = self._redact_exception_with_context_or_cause(
+                    exception.__cause__, visited
+                )
+        return exception
 
     def filter(self, record) -> bool:
         if not self.is_log_masking_enabled():
@@ -288,7 +311,7 @@ def filter(self, record) -> bool:
                     record.__dict__[k] = self.redact(v)
             if record.exc_info and record.exc_info[1] is not None:
                 exc = record.exc_info[1]
-                self._redact_exception_with_context(exc)
+                self._redact_exception_with_context_or_cause(exc)
         record.__dict__[self.ALREADY_FILTERED_FLAG] = True
 
         return True