| 40 | |
| 41 | |
| 42 | class UnsubscribeEncoder: |
| 43 | @staticmethod |
| 44 | def encode( |
| 45 | action: UnsubscribeAction, |
| 46 | data: Union[int, UnsubscribeOriginalData], |
| 47 | force_web: bool = False, |
| 48 | ) -> UnsubscribeLink: |
| 49 | if config.UNSUBSCRIBER and not force_web: |
| 50 | return UnsubscribeLink(UnsubscribeEncoder.encode_mailto(action, data), True) |
| 51 | return UnsubscribeLink(UnsubscribeEncoder.encode_url(action, data), False) |
| 52 | |
| 53 | @classmethod |
| 54 | def encode_subject( |
| 55 | cls, action: UnsubscribeAction, data: Union[int, UnsubscribeOriginalData] |
| 56 | ) -> str: |
| 57 | if action != UnsubscribeAction.OriginalUnsubscribeMailto and not isinstance( |
| 58 | data, int |
| 59 | ): |
| 60 | raise ValueError(f"Data has to be an int for an action of type {action}") |
| 61 | if action == UnsubscribeAction.OriginalUnsubscribeMailto: |
| 62 | if type(data) is not UnsubscribeOriginalData: |
| 63 | raise ValueError( |
| 64 | f"Data has to be an UnsubscribeOriginalData for an action of type {action}" |
| 65 | ) |
| 66 | # Initial 0 is the version number. If we need to add support for extra use-cases we can bump up this number |
| 67 | data = (0, data.alias_id, data.recipient, data.subject) |
| 68 | payload = (action.value, data) |
| 69 | serialized_data = ( |
| 70 | base64.urlsafe_b64encode(json.dumps(payload).encode("utf-8")) |
| 71 | .rstrip(b"=") |
| 72 | .decode("utf-8") |
| 73 | ) |
| 74 | signed_data = cls._get_signer().sign(serialized_data).decode("utf-8") |
| 75 | encoded_request = f"{UNSUB_PREFIX}.{signed_data}" |
| 76 | if len(encoded_request) > 512: |
| 77 | LOG.w("Encoded request is longer than 512 chars") |
| 78 | return encoded_request |
| 79 | |
| 80 | @staticmethod |
| 81 | def encode_mailto( |
| 82 | action: UnsubscribeAction, data: Union[int, UnsubscribeOriginalData] |
| 83 | ) -> str: |
| 84 | subject = UnsubscribeEncoder.encode_subject(action, data) |
| 85 | return f"mailto:{config.UNSUBSCRIBER}?subject={subject}" |
| 86 | |
| 87 | @staticmethod |
| 88 | def encode_url( |
| 89 | action: UnsubscribeAction, data: Union[int, UnsubscribeOriginalData] |
| 90 | ) -> str: |
| 91 | if action == UnsubscribeAction.DisableAlias: |
| 92 | return f"{config.URL}/dashboard/unsubscribe/{data}" |
| 93 | if action == UnsubscribeAction.DisableContact: |
| 94 | return f"{config.URL}/dashboard/block_contact/{data}" |
| 95 | if action in ( |
| 96 | UnsubscribeAction.UnsubscribeNewsletter, |
| 97 | UnsubscribeAction.OriginalUnsubscribeMailto, |
| 98 | ): |
| 99 | encoded = UnsubscribeEncoder.encode_subject(action, data) |
nothing calls this directly
no outgoing calls
no test coverage detected