diff --git a/src/sentry/sentry_apps/api/serializers/app_platform_event.py b/src/sentry/sentry_apps/api/serializers/app_platform_event.py index 6e6124e3a53d..b02060242e10 100644 --- a/src/sentry/sentry_apps/api/serializers/app_platform_event.py +++ b/src/sentry/sentry_apps/api/serializers/app_platform_event.py @@ -5,9 +5,9 @@ from typing import Any, TypedDict from uuid import uuid4 -from sentry.sentry_apps.models.sentry_app import MASKED_VALUE from sentry.sentry_apps.models.sentry_app_installation import SentryAppInstallation from sentry.sentry_apps.services.app.model import RpcSentryAppInstallation +from sentry.sentry_apps.utils.headers import mask_header_values, parse_custom_headers from sentry.sentry_apps.utils.webhooks import SentryAppActionType, SentryAppResourceType from sentry.users.models.user import User from sentry.users.services.user import RpcUser @@ -111,12 +111,7 @@ def sentry_headers(self) -> dict[str, str]: @property def custom_headers(self) -> dict[str, str]: """User-configured headers parsed from the SentryApp's webhook_headers.""" - headers: dict[str, str] = {} - for header in self.install.sentry_app.webhook_headers or []: - name, separator, value = header.partition(":") - if separator: - headers[name.strip()] = value.strip() - return headers + return parse_custom_headers(self.install.sentry_app.webhook_headers) @property def headers(self) -> dict[str, str]: @@ -132,7 +127,7 @@ def masked_custom_headers(self) -> dict[str, str]: never persisted to the request buffer. The names are kept so the debug UI can show which custom headers were sent without leaking the values. """ - return {name: MASKED_VALUE for name in self.custom_headers} + return mask_header_values(self.custom_headers) @property def loggable_headers(self) -> dict[str, str]: diff --git a/src/sentry/sentry_apps/utils/headers.py b/src/sentry/sentry_apps/utils/headers.py new file mode 100644 index 000000000000..6ec0464363bd --- /dev/null +++ b/src/sentry/sentry_apps/utils/headers.py @@ -0,0 +1,16 @@ +from collections.abc import Mapping, Sequence + +from sentry.sentry_apps.models.sentry_app import MASKED_VALUE + + +def parse_custom_headers(webhook_headers: Sequence[str]) -> dict[str, str]: + headers: dict[str, str] = {} + for header in webhook_headers: + name, separator, value = header.partition(":") + if separator: + headers[name.strip()] = value.strip() + return headers + + +def mask_header_values(headers: Mapping[str, str]) -> dict[str, str]: + return {name: MASKED_VALUE for name in headers}