| 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( |