1 files changed, 201 insertions, 0 deletions
diff --git a/mastodon/push.py b/mastodon/push.py
new file mode 100644
index 0000000..def348a
--- /dev/null
+++ b/mastodon/push.py
@@ -0,0 +1,201 @@
+# push.py - webpush endpoints and tooling
+import base64
+import os
+import json
+from .versions import _DICT_VERSION_PUSH, _DICT_VERSION_PUSH_NOTIF
+from .errors import MastodonIllegalArgumentError
+from .utility import api_version
+from .compat import IMPL_HAS_CRYPTO, ec, serialization, default_backend
+from .compat import IMPL_HAS_ECE, http_ece
+from .internals import Mastodon as Internals
+class Mastodon(Internals):
+    ###
+    # Reading data: Webpush subscriptions
+    ###
+    @api_version("2.4.0", "2.4.0", _DICT_VERSION_PUSH)
+    def push_subscription(self):
+        """
+        Fetch the current push subscription the logged-in user has for this app.
+        Returns a :ref:`push subscription dict <push subscription dict>`.
+        """
+        return self.__api_request('GET', '/api/v1/push/subscription')
+    ###
+    # Writing data: Push subscriptions
+    ###
+    @api_version("2.4.0", "2.4.0", _DICT_VERSION_PUSH)
+    def push_subscription_set(self, endpoint, encrypt_params, follow_events=None,
+                              favourite_events=None, reblog_events=None,
+                              mention_events=None, poll_events=None,
+                              follow_request_events=None, status_events=None, policy='all'):
+        """
+        Sets up or modifies the push subscription the logged-in user has for this app.
+        `endpoint` is the endpoint URL mastodon should call for pushes. Note that mastodon
+        requires https for this URL. `encrypt_params` is a dict with key parameters that allow
+        the server to encrypt data for you: A public key `pubkey` and a shared secret `auth`.
+        You can generate this as well as the corresponding private key using the
+        :ref:`push_subscription_generate_keys() <push_subscription_generate_keys()>` function.
+        `policy` controls what sources will generate webpush events. Valid values are
+        `all`, `none`, `follower` and `followed`.
+        The rest of the parameters controls what kind of events you wish to subscribe to.
+        Returns a :ref:`push subscription dict <push subscription dict>`.
+        """
+        if not policy in ['all', 'none', 'follower', 'followed']:
+            raise MastodonIllegalArgumentError("Valid values for policy are 'all', 'none', 'follower' or 'followed'.")
+        endpoint = Mastodon.__protocolize(endpoint)
+        push_pubkey_b64 = base64.b64encode(encrypt_params['pubkey'])
+        push_auth_b64 = base64.b64encode(encrypt_params['auth'])
+        params = {
+            'subscription[endpoint]': endpoint,
+            'subscription[keys][p256dh]': push_pubkey_b64,
+            'subscription[keys][auth]': push_auth_b64,
+            'policy': policy
+        }
+        if follow_events is not None:
+            params['data[alerts][follow]'] = follow_events
+        if favourite_events is not None:
+            params['data[alerts][favourite]'] = favourite_events
+        if reblog_events is not None:
+            params['data[alerts][reblog]'] = reblog_events
+        if mention_events is not None:
+            params['data[alerts][mention]'] = mention_events
+        if poll_events is not None:
+            params['data[alerts][poll]'] = poll_events
+        if follow_request_events is not None:
+            params['data[alerts][follow_request]'] = follow_request_events
+        if follow_request_events is not None:
+            params['data[alerts][status]'] = status_events
+        # Canonicalize booleans
+        params = self.__generate_params(params)
+        return self.__api_request('POST', '/api/v1/push/subscription', params)
+    @api_version("2.4.0", "2.4.0", _DICT_VERSION_PUSH)
+    def push_subscription_update(self, follow_events=None,
+                                 favourite_events=None, reblog_events=None,
+                                 mention_events=None, poll_events=None,
+                                 follow_request_events=None):
+        """
+        Modifies what kind of events the app wishes to subscribe to.
+        Returns the updated :ref:`push subscription dict <push subscription dict>`.
+        """
+        params = {}
+        if follow_events is not None:
+            params['data[alerts][follow]'] = follow_events
+        if favourite_events is not None:
+            params['data[alerts][favourite]'] = favourite_events
+        if reblog_events is not None:
+            params['data[alerts][reblog]'] = reblog_events
+        if mention_events is not None:
+            params['data[alerts][mention]'] = mention_events
+        if poll_events is not None:
+            params['data[alerts][poll]'] = poll_events
+        if follow_request_events is not None:
+            params['data[alerts][follow_request]'] = follow_request_events
+        # Canonicalize booleans
+        params = self.__generate_params(params)
+        return self.__api_request('PUT', '/api/v1/push/subscription', params)
+    @api_version("2.4.0", "2.4.0", "2.4.0")
+    def push_subscription_delete(self):
+        """
+        Remove the current push subscription the logged-in user has for this app.
+        """
+        self.__api_request('DELETE', '/api/v1/push/subscription')
+    ###
+    # Push subscription crypto utilities
+    ###
+    def push_subscription_generate_keys(self):
+        """
+        Generates a private key, public key and shared secret for use in webpush subscriptions.
+        Returns two dicts: One with the private key and shared secret and another with the
+        public key and shared secret.
+        """
+        if not IMPL_HAS_CRYPTO:
+            raise NotImplementedError(
+                'To use the crypto tools, please install the webpush feature dependencies.')
+        push_key_pair = ec.generate_private_key(ec.SECP256R1(), default_backend())
+        push_key_priv = push_key_pair.private_numbers().private_value
+        try: 
+            push_key_pub = push_key_pair.public_key().public_bytes(
+                serialization.Encoding.X962,
+                serialization.PublicFormat.UncompressedPoint,
+            )
+        except:
+            push_key_pub = push_key_pair.public_key().public_numbers().encode_point()
+            
+        push_shared_secret = os.urandom(16)
+        priv_dict = {
+            'privkey': push_key_priv,
+            'auth': push_shared_secret
+        }
+        pub_dict = {
+            'pubkey': push_key_pub,
+            'auth': push_shared_secret
+        }
+        return priv_dict, pub_dict
+    @api_version("2.4.0", "2.4.0", _DICT_VERSION_PUSH_NOTIF)
+    def push_subscription_decrypt_push(self, data, decrypt_params, encryption_header, crypto_key_header):
+        """
+        Decrypts `data` received in a webpush request. Requires the private key dict
+        from :ref:`push_subscription_generate_keys() <push_subscription_generate_keys()>` (`decrypt_params`) as well as the
+        Encryption and server Crypto-Key headers from the received webpush
+        Returns the decoded webpush as a :ref:`push notification dict <push notification dict>`.
+        """
+        if (not IMPL_HAS_ECE) or (not IMPL_HAS_CRYPTO):
+            raise NotImplementedError(
+                'To use the crypto tools, please install the webpush feature dependencies.')
+        salt = self.__decode_webpush_b64(encryption_header.split("salt=")[1].strip())
+        dhparams = self.__decode_webpush_b64(crypto_key_header.split("dh=")[1].split(";")[0].strip())
+        p256ecdsa = self.__decode_webpush_b64(crypto_key_header.split("p256ecdsa=")[1].strip())
+        dec_key = ec.derive_private_key(decrypt_params['privkey'], ec.SECP256R1(), default_backend())
+        decrypted = http_ece.decrypt(
+            data,
+            salt=salt,
+            key=p256ecdsa,
+            private_key=dec_key,
+            dh=dhparams,
+            auth_secret=decrypt_params['auth'],
+            keylabel="P-256",
+            version="aesgcm"
+        )
+        return json.loads(decrypted.decode('utf-8'), object_hook=Mastodon.__json_hooks)

diff --git a/mastodon/push.py b/mastodon/push.py new file mode 100644 index 0000000..def348a --- /dev/null +++ b/mastodon/push.py
@@ -0,0 +1,201 @@
	1	# push.py - webpush endpoints and tooling
	2
	3	import base64
	4	import os
	5	import json
	6
	7	from .versions import _DICT_VERSION_PUSH, _DICT_VERSION_PUSH_NOTIF
	8	from .errors import MastodonIllegalArgumentError
	9	from .utility import api_version
	10	from .compat import IMPL_HAS_CRYPTO, ec, serialization, default_backend
	11	from .compat import IMPL_HAS_ECE, http_ece
	12
	13	from .internals import Mastodon as Internals
	14
	15	class Mastodon(Internals):
	16	###
	17	# Reading data: Webpush subscriptions
	18	###
	19	@api_version("2.4.0", "2.4.0", _DICT_VERSION_PUSH)
	20	def push_subscription(self):
	21	"""
	22	Fetch the current push subscription the logged-in user has for this app.
	23
	24	Returns a :ref:`push subscription dict <push subscription dict>`.
	25	"""
	26	return self.__api_request('GET', '/api/v1/push/subscription')
	27
	28	###
	29	# Writing data: Push subscriptions
	30	###
	31	@api_version("2.4.0", "2.4.0", _DICT_VERSION_PUSH)
	32	def push_subscription_set(self, endpoint, encrypt_params, follow_events=None,
	33	favourite_events=None, reblog_events=None,
	34	mention_events=None, poll_events=None,
	35	follow_request_events=None, status_events=None, policy='all'):
	36	"""
	37	Sets up or modifies the push subscription the logged-in user has for this app.
	38
	39	`endpoint` is the endpoint URL mastodon should call for pushes. Note that mastodon
	40	requires https for this URL. `encrypt_params` is a dict with key parameters that allow
	41	the server to encrypt data for you: A public key `pubkey` and a shared secret `auth`.
	42	You can generate this as well as the corresponding private key using the
	43	:ref:`push_subscription_generate_keys() <push_subscription_generate_keys()>` function.
	44
	45	`policy` controls what sources will generate webpush events. Valid values are
	46	`all`, `none`, `follower` and `followed`.
	47
	48	The rest of the parameters controls what kind of events you wish to subscribe to.
	49
	50	Returns a :ref:`push subscription dict <push subscription dict>`.
	51	"""
	52	if not policy in ['all', 'none', 'follower', 'followed']:
	53	raise MastodonIllegalArgumentError("Valid values for policy are 'all', 'none', 'follower' or 'followed'.")
	54
	55	endpoint = Mastodon.__protocolize(endpoint)
	56
	57	push_pubkey_b64 = base64.b64encode(encrypt_params['pubkey'])
	58	push_auth_b64 = base64.b64encode(encrypt_params['auth'])
	59
	60	params = {
	61	'subscription[endpoint]': endpoint,
	62	'subscription[keys][p256dh]': push_pubkey_b64,
	63	'subscription[keys][auth]': push_auth_b64,
	64	'policy': policy
	65	}
	66
	67	if follow_events is not None:
	68	params['data[alerts][follow]'] = follow_events
	69
	70	if favourite_events is not None:
	71	params['data[alerts][favourite]'] = favourite_events
	72
	73	if reblog_events is not None:
	74	params['data[alerts][reblog]'] = reblog_events
	75
	76	if mention_events is not None:
	77	params['data[alerts][mention]'] = mention_events
	78
	79	if poll_events is not None:
	80	params['data[alerts][poll]'] = poll_events
	81
	82	if follow_request_events is not None:
	83	params['data[alerts][follow_request]'] = follow_request_events
	84
	85	if follow_request_events is not None:
	86	params['data[alerts][status]'] = status_events
	87
	88	# Canonicalize booleans
	89	params = self.__generate_params(params)
	90
	91	return self.__api_request('POST', '/api/v1/push/subscription', params)
	92
	93	@api_version("2.4.0", "2.4.0", _DICT_VERSION_PUSH)
	94	def push_subscription_update(self, follow_events=None,
	95	favourite_events=None, reblog_events=None,
	96	mention_events=None, poll_events=None,
	97	follow_request_events=None):
	98	"""
	99	Modifies what kind of events the app wishes to subscribe to.
	100
	101	Returns the updated :ref:`push subscription dict <push subscription dict>`.
	102	"""
	103	params = {}
	104
	105	if follow_events is not None:
	106	params['data[alerts][follow]'] = follow_events
	107
	108	if favourite_events is not None:
	109	params['data[alerts][favourite]'] = favourite_events
	110
	111	if reblog_events is not None:
	112	params['data[alerts][reblog]'] = reblog_events
	113
	114	if mention_events is not None:
	115	params['data[alerts][mention]'] = mention_events
	116
	117	if poll_events is not None:
	118	params['data[alerts][poll]'] = poll_events
	119
	120	if follow_request_events is not None:
	121	params['data[alerts][follow_request]'] = follow_request_events
	122
	123	# Canonicalize booleans
	124	params = self.__generate_params(params)
	125
	126	return self.__api_request('PUT', '/api/v1/push/subscription', params)
	127
	128	@api_version("2.4.0", "2.4.0", "2.4.0")
	129	def push_subscription_delete(self):
	130	"""
	131	Remove the current push subscription the logged-in user has for this app.
	132	"""
	133	self.__api_request('DELETE', '/api/v1/push/subscription')
	134
	135	###
	136	# Push subscription crypto utilities
	137	###
	138	def push_subscription_generate_keys(self):
	139	"""
	140	Generates a private key, public key and shared secret for use in webpush subscriptions.
	141
	142	Returns two dicts: One with the private key and shared secret and another with the
	143	public key and shared secret.
	144	"""
	145	if not IMPL_HAS_CRYPTO:
	146	raise NotImplementedError(
	147	'To use the crypto tools, please install the webpush feature dependencies.')
	148
	149	push_key_pair = ec.generate_private_key(ec.SECP256R1(), default_backend())
	150	push_key_priv = push_key_pair.private_numbers().private_value
	151	try:
	152	push_key_pub = push_key_pair.public_key().public_bytes(
	153	serialization.Encoding.X962,
	154	serialization.PublicFormat.UncompressedPoint,
	155	)
	156	except:
	157	push_key_pub = push_key_pair.public_key().public_numbers().encode_point()
	158
	159	push_shared_secret = os.urandom(16)
	160
	161	priv_dict = {
	162	'privkey': push_key_priv,
	163	'auth': push_shared_secret
	164	}
	165
	166	pub_dict = {
	167	'pubkey': push_key_pub,
	168	'auth': push_shared_secret
	169	}
	170
	171	return priv_dict, pub_dict
	172
	173	@api_version("2.4.0", "2.4.0", _DICT_VERSION_PUSH_NOTIF)
	174	def push_subscription_decrypt_push(self, data, decrypt_params, encryption_header, crypto_key_header):
	175	"""
	176	Decrypts `data` received in a webpush request. Requires the private key dict
	177	from :ref:`push_subscription_generate_keys() <push_subscription_generate_keys()>` (`decrypt_params`) as well as the
	178	Encryption and server Crypto-Key headers from the received webpush
	179
	180	Returns the decoded webpush as a :ref:`push notification dict <push notification dict>`.
	181	"""
	182	if (not IMPL_HAS_ECE) or (not IMPL_HAS_CRYPTO):
	183	raise NotImplementedError(
	184	'To use the crypto tools, please install the webpush feature dependencies.')
	185
	186	salt = self.__decode_webpush_b64(encryption_header.split("salt=")[1].strip())
	187	dhparams = self.__decode_webpush_b64(crypto_key_header.split("dh=")[1].split(";")[0].strip())
	188	p256ecdsa = self.__decode_webpush_b64(crypto_key_header.split("p256ecdsa=")[1].strip())
	189	dec_key = ec.derive_private_key(decrypt_params['privkey'], ec.SECP256R1(), default_backend())
	190	decrypted = http_ece.decrypt(
	191	data,
	192	salt=salt,
	193	key=p256ecdsa,
	194	private_key=dec_key,
	195	dh=dhparams,
	196	auth_secret=decrypt_params['auth'],
	197	keylabel="P-256",
	198	version="aesgcm"
	199	)
	200
	201	return json.loads(decrypted.decode('utf-8'), object_hook=Mastodon.__json_hooks)