|
| 1 | +"""Producer-side translation of AdCP universal macros in pixel URLs. |
| 2 | +
|
| 3 | +``native`` mappings are a deliberate raw-token escape hatch for downstream ad |
| 4 | +servers. They bypass percent-encoding, so this module validates every native |
| 5 | +entry before translating the URL, including entries the URL does not use. |
| 6 | +""" |
| 7 | + |
| 8 | +from __future__ import annotations |
| 9 | + |
| 10 | +import re |
| 11 | +from collections.abc import Mapping |
| 12 | +from dataclasses import dataclass |
| 13 | +from typing import Literal, TypeAlias |
| 14 | + |
| 15 | +_UNIVERSAL_MACRO = re.compile(r"\{[A-Z][A-Z0-9_]*\}") |
| 16 | +_NATIVE_TOKEN_SHAPE = re.compile( |
| 17 | + r"(?:%%[^\n\r\u2028\u2029]+%%|" |
| 18 | + r"\{\{[^\n\r\u2028\u2029]+\}\}|" |
| 19 | + r"\$\{[^\n\r\u2028\u2029]+\}|" |
| 20 | + r"\[[A-Z][A-Z0-9_]*\])" |
| 21 | +) |
| 22 | +_CONSENT_MACROS = frozenset( |
| 23 | + { |
| 24 | + "{GDPR}", |
| 25 | + "{GDPR_CONSENT}", |
| 26 | + "{US_PRIVACY}", |
| 27 | + "{GPP_STRING}", |
| 28 | + "{GPP_SID}", |
| 29 | + "{LIMIT_AD_TRACKING}", |
| 30 | + } |
| 31 | +) |
| 32 | +_UNRESERVED_BYTES = frozenset(b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-._~") |
| 33 | + |
| 34 | + |
| 35 | +@dataclass(frozen=True, slots=True) |
| 36 | +class NativeMacroMapping: |
| 37 | + """A downstream ad-server token inserted without percent-encoding.""" |
| 38 | + |
| 39 | + native: str |
| 40 | + |
| 41 | + |
| 42 | +@dataclass(frozen=True, slots=True) |
| 43 | +class ValueMacroMapping: |
| 44 | + """A literal value encoded with the RFC 3986 unreserved whitelist.""" |
| 45 | + |
| 46 | + value: str |
| 47 | + |
| 48 | + |
| 49 | +MacroMappingEntry: TypeAlias = NativeMacroMapping | ValueMacroMapping |
| 50 | +MacroMapping: TypeAlias = Mapping[str, MacroMappingEntry] |
| 51 | +UniversalMacroTranslationErrorCode: TypeAlias = Literal["unsafe_native_mapping"] |
| 52 | + |
| 53 | + |
| 54 | +@dataclass(slots=True) |
| 55 | +class TranslateUniversalMacrosResult: |
| 56 | + """Translated URL and deterministic diagnostics. |
| 57 | +
|
| 58 | + ``dropped_params`` preserves query-parameter occurrence order and may |
| 59 | + contain duplicate keys. All macro diagnostic lists are deduplicated. |
| 60 | + URL-scoped diagnostics use first query occurrence order; mapping-scoped |
| 61 | + diagnostics preserve mapping iteration order. |
| 62 | + """ |
| 63 | + |
| 64 | + url: str |
| 65 | + dropped_params: list[str] |
| 66 | + unmapped_macros: list[str] |
| 67 | + dropped_consent_macros: list[str] |
| 68 | + frozen_consent_macros: list[str] |
| 69 | + suspect_native_values: list[str] |
| 70 | + |
| 71 | + |
| 72 | +class UniversalMacroTranslationError(ValueError): |
| 73 | + """Typed rejection raised before an unsafe native token can be emitted.""" |
| 74 | + |
| 75 | + code: UniversalMacroTranslationErrorCode |
| 76 | + macro: str |
| 77 | + |
| 78 | + def __init__(self, macro: str) -> None: |
| 79 | + self.code = "unsafe_native_mapping" |
| 80 | + self.macro = macro |
| 81 | + super().__init__(f"native mapping for {macro!r} contains an unsafe control character") |
| 82 | + |
| 83 | + |
| 84 | +def encode_unreserved(raw: str) -> str: |
| 85 | + """UTF-8 encode ``raw``, escaping every byte outside RFC 3986 unreserved. |
| 86 | +
|
| 87 | + Percent escapes use uppercase hexadecimal. Unlike |
| 88 | + :func:`urllib.parse.quote`, this helper does not preserve ``/``. |
| 89 | + """ |
| 90 | + |
| 91 | + return "".join( |
| 92 | + chr(byte) if byte in _UNRESERVED_BYTES else f"%{byte:02X}" for byte in raw.encode("utf-8") |
| 93 | + ) |
| 94 | + |
| 95 | + |
| 96 | +def _has_unsafe_native_character(value: str) -> bool: |
| 97 | + return any(ord(character) <= 0x1F or ord(character) == 0x7F for character in value) |
| 98 | + |
| 99 | + |
| 100 | +def _append_once(items: list[str], seen: set[str], value: str) -> None: |
| 101 | + if value not in seen: |
| 102 | + seen.add(value) |
| 103 | + items.append(value) |
| 104 | + |
| 105 | + |
| 106 | +def translate_universal_macros( |
| 107 | + pixel_url: str, |
| 108 | + mapping: MacroMapping, |
| 109 | +) -> TranslateUniversalMacrosResult: |
| 110 | + """Translate universal macros in query-parameter values. |
| 111 | +
|
| 112 | + ``ValueMacroMapping`` values are UTF-8 percent-encoded with |
| 113 | + :func:`encode_unreserved`; ``NativeMacroMapping`` values are inserted |
| 114 | + verbatim after the full mapping passes the control-character guard. If a |
| 115 | + parameter value contains any unmapped universal macro, that whole parameter |
| 116 | + is dropped. Query keys, the path, the fragment, and parameters without |
| 117 | + universal macros pass through byte-for-byte. Replacement is single-pass. |
| 118 | +
|
| 119 | + Consent macros supplied through ``ValueMacroMapping`` are translated but |
| 120 | + reported in ``frozen_consent_macros`` because freezing impression-time |
| 121 | + consent at producer time can create a privacy defect. Callers should also |
| 122 | + inspect ``dropped_consent_macros`` and ``suspect_native_values`` before |
| 123 | + publishing a tracker. |
| 124 | +
|
| 125 | + Raises: |
| 126 | + UniversalMacroTranslationError: A native mapping, used or unused, |
| 127 | + contains U+0000-U+001F or U+007F. No URL is emitted. |
| 128 | + TypeError: A mapping entry is not a supported typed mapping model. |
| 129 | + """ |
| 130 | + |
| 131 | + frozen_consent_macros: list[str] = [] |
| 132 | + frozen_seen: set[str] = set() |
| 133 | + suspect_native_values: list[str] = [] |
| 134 | + suspect_seen: set[str] = set() |
| 135 | + validated_mapping: dict[str, MacroMappingEntry] = {} |
| 136 | + |
| 137 | + # Validate the entire raw-token trust boundary before doing any URL work. |
| 138 | + # An unused unsafe entry must reject just like an entry present in the URL. |
| 139 | + for macro, entry in mapping.items(): |
| 140 | + if isinstance(entry, NativeMacroMapping): |
| 141 | + native = entry.native |
| 142 | + if _has_unsafe_native_character(native): |
| 143 | + raise UniversalMacroTranslationError(macro) |
| 144 | + validated_mapping[macro] = NativeMacroMapping(native=native) |
| 145 | + elif isinstance(entry, ValueMacroMapping): |
| 146 | + value = entry.value |
| 147 | + if macro in _CONSENT_MACROS: |
| 148 | + _append_once(frozen_consent_macros, frozen_seen, macro) |
| 149 | + if _NATIVE_TOKEN_SHAPE.fullmatch(value): |
| 150 | + _append_once(suspect_native_values, suspect_seen, macro) |
| 151 | + validated_mapping[macro] = ValueMacroMapping(value=value) |
| 152 | + else: |
| 153 | + raise TypeError( |
| 154 | + f"mapping entry for {macro!r} must be NativeMacroMapping " "or ValueMacroMapping" |
| 155 | + ) |
| 156 | + |
| 157 | + fragment_index = pixel_url.find("#") |
| 158 | + if fragment_index == -1: |
| 159 | + without_fragment = pixel_url |
| 160 | + fragment = "" |
| 161 | + else: |
| 162 | + without_fragment = pixel_url[:fragment_index] |
| 163 | + fragment = pixel_url[fragment_index:] |
| 164 | + |
| 165 | + query_index = without_fragment.find("?") |
| 166 | + if query_index == -1: |
| 167 | + return TranslateUniversalMacrosResult( |
| 168 | + url=pixel_url, |
| 169 | + dropped_params=[], |
| 170 | + unmapped_macros=[], |
| 171 | + dropped_consent_macros=[], |
| 172 | + frozen_consent_macros=frozen_consent_macros, |
| 173 | + suspect_native_values=suspect_native_values, |
| 174 | + ) |
| 175 | + |
| 176 | + base = without_fragment[:query_index] |
| 177 | + raw_query = without_fragment[query_index + 1 :] |
| 178 | + |
| 179 | + dropped_params: list[str] = [] |
| 180 | + unmapped_macros: list[str] = [] |
| 181 | + unmapped_seen: set[str] = set() |
| 182 | + dropped_consent_macros: list[str] = [] |
| 183 | + dropped_consent_seen: set[str] = set() |
| 184 | + output_parts: list[str] = [] |
| 185 | + |
| 186 | + for raw_param in raw_query.split("&"): |
| 187 | + key, separator, value = raw_param.partition("=") |
| 188 | + tokens = _UNIVERSAL_MACRO.findall(value) |
| 189 | + if not tokens: |
| 190 | + output_parts.append(raw_param) |
| 191 | + continue |
| 192 | + |
| 193 | + missing = [token for token in tokens if token not in validated_mapping] |
| 194 | + if missing: |
| 195 | + dropped_params.append(key) |
| 196 | + for macro in missing: |
| 197 | + _append_once(unmapped_macros, unmapped_seen, macro) |
| 198 | + if macro in _CONSENT_MACROS: |
| 199 | + _append_once( |
| 200 | + dropped_consent_macros, |
| 201 | + dropped_consent_seen, |
| 202 | + macro, |
| 203 | + ) |
| 204 | + continue |
| 205 | + |
| 206 | + def replace(match: re.Match[str]) -> str: |
| 207 | + macro = match.group(0) |
| 208 | + entry = validated_mapping[macro] |
| 209 | + if isinstance(entry, NativeMacroMapping): |
| 210 | + return entry.native |
| 211 | + if isinstance(entry, ValueMacroMapping): |
| 212 | + return encode_unreserved(entry.value) |
| 213 | + # The mapping-wide validation above makes this unreachable even |
| 214 | + # for mutable custom Mapping implementations under normal use. |
| 215 | + raise TypeError(f"unsupported mapping entry for {macro!r}") |
| 216 | + |
| 217 | + translated = _UNIVERSAL_MACRO.sub(replace, value) |
| 218 | + output_parts.append(f"{key}{separator}{translated}") |
| 219 | + |
| 220 | + new_query = "&".join(output_parts) |
| 221 | + url = f"{base}?{new_query}{fragment}" if new_query else f"{base}{fragment}" |
| 222 | + return TranslateUniversalMacrosResult( |
| 223 | + url=url, |
| 224 | + dropped_params=dropped_params, |
| 225 | + unmapped_macros=unmapped_macros, |
| 226 | + dropped_consent_macros=dropped_consent_macros, |
| 227 | + frozen_consent_macros=frozen_consent_macros, |
| 228 | + suspect_native_values=suspect_native_values, |
| 229 | + ) |
| 230 | + |
| 231 | + |
| 232 | +__all__ = [ |
| 233 | + "MacroMapping", |
| 234 | + "MacroMappingEntry", |
| 235 | + "NativeMacroMapping", |
| 236 | + "TranslateUniversalMacrosResult", |
| 237 | + "UniversalMacroTranslationError", |
| 238 | + "UniversalMacroTranslationErrorCode", |
| 239 | + "ValueMacroMapping", |
| 240 | + "encode_unreserved", |
| 241 | + "translate_universal_macros", |
| 242 | +] |
0 commit comments