From d77c709b4e7452e50aa221b23241c836a7be4496 Mon Sep 17 00:00:00 2001 From: Ayush Saini <36878972+imayushsaini@users.noreply.github.com> Date: Sat, 24 Jun 2023 18:14:24 +0530 Subject: [PATCH] added py vapid --- .../python-site-packages/py_vapid/__init__.py | 359 ++++++++++++++++++ .../python-site-packages/py_vapid/jwt.py | 87 +++++ .../python-site-packages/py_vapid/main.py | 115 ++++++ .../py_vapid/tests/test_vapid.py | 282 ++++++++++++++ .../python-site-packages/py_vapid/utils.py | 39 ++ 5 files changed, 882 insertions(+) create mode 100644 dist/ba_data/python-site-packages/py_vapid/__init__.py create mode 100644 dist/ba_data/python-site-packages/py_vapid/jwt.py create mode 100644 dist/ba_data/python-site-packages/py_vapid/main.py create mode 100644 dist/ba_data/python-site-packages/py_vapid/tests/test_vapid.py create mode 100644 dist/ba_data/python-site-packages/py_vapid/utils.py diff --git a/dist/ba_data/python-site-packages/py_vapid/__init__.py b/dist/ba_data/python-site-packages/py_vapid/__init__.py new file mode 100644 index 0000000..76edc62 --- /dev/null +++ b/dist/ba_data/python-site-packages/py_vapid/__init__.py @@ -0,0 +1,359 @@ +# This Source Code Form is subject to the terms of the Mozilla Public +# License, v. 2.0. If a copy of the MPL was not distributed with this +# file, You can obtain one at http://mozilla.org/MPL/2.0/. + +import os +import logging +import binascii +import time +import re +import copy + +from cryptography.hazmat.backends import default_backend +from cryptography.hazmat.primitives.asymmetric import ec, utils as ecutils +from cryptography.hazmat.primitives import serialization + +from cryptography.hazmat.primitives import hashes +from cryptography.exceptions import InvalidSignature + +from py_vapid.utils import b64urldecode, b64urlencode +from py_vapid.jwt import sign + +# Show compliance version. For earlier versions see previously tagged releases. +VERSION = "VAPID-RFC/ECE-RFC" + + +class VapidException(Exception): + """An exception wrapper for Vapid.""" + pass + + +class Vapid01(object): + """Minimal VAPID Draft 01 signature generation library. + + https://tools.ietf.org/html/draft-ietf-webpush-vapid-01 + + """ + _private_key = None + _public_key = None + _schema = "WebPush" + + def __init__(self, private_key=None, conf=None): + """Initialize VAPID with an optional private key. + + :param private_key: A private key object + :type private_key: ec.EllipticCurvePrivateKey + + """ + if conf is None: + conf = {} + self.conf = conf + self.private_key = private_key + if private_key: + self._public_key = self.private_key.public_key() + + @classmethod + def from_raw(cls, private_raw): + """Initialize VAPID using a private key point in "raw" or + "uncompressed" form. Raw keys consist of a single, 32 octet + encoded integer. + + :param private_raw: A private key point in uncompressed form. + :type private_raw: bytes + + """ + key = ec.derive_private_key( + int(binascii.hexlify(b64urldecode(private_raw)), 16), + curve=ec.SECP256R1(), + backend=default_backend()) + return cls(key) + + @classmethod + def from_raw_public(cls, public_raw): + key = ec.EllipticCurvePublicKey.from_encoded_point( + curve=ec.SECP256R1(), + data=b64urldecode(public_raw) + ) + ss = cls() + ss._public_key = key + return ss + + @classmethod + def from_pem(cls, private_key): + """Initialize VAPID using a private key in PEM format. + + :param private_key: A private key in PEM format. + :type private_key: bytes + + """ + # not sure why, but load_pem_private_key fails to deserialize + return cls.from_der( + b''.join(private_key.splitlines()[1:-1])) + + @classmethod + def from_der(cls, private_key): + """Initialize VAPID using a private key in DER format. + + :param private_key: A private key in DER format and Base64-encoded. + :type private_key: bytes + + """ + key = serialization.load_der_private_key(b64urldecode(private_key), + password=None, + backend=default_backend()) + return cls(key) + + @classmethod + def from_file(cls, private_key_file=None): + """Initialize VAPID using a file containing a private key in PEM or + DER format. + + :param private_key_file: Name of the file containing the private key + :type private_key_file: str + + """ + if not os.path.isfile(private_key_file): + logging.info("Private key not found, generating key...") + vapid = cls() + vapid.generate_keys() + vapid.save_key(private_key_file) + return vapid + with open(private_key_file, 'r') as file: + private_key = file.read() + try: + if "-----BEGIN" in private_key: + vapid = cls.from_pem(private_key.encode('utf8')) + else: + vapid = cls.from_der(private_key.encode('utf8')) + return vapid + except Exception as exc: + logging.error("Could not open private key file: %s", repr(exc)) + raise VapidException(exc) + + @classmethod + def from_string(cls, private_key): + """Initialize VAPID using a string containing the private key. This + will try to determine if the key is in RAW or DER format. + + :param private_key: String containing the key info + :type private_key: str + + """ + + pkey = private_key.encode().replace(b"\n", b"") + key = b64urldecode(pkey) + if len(key) == 32: + return cls.from_raw(pkey) + return cls.from_der(pkey) + + @classmethod + def verify(cls, key, auth): + """Verify a VAPID authorization token. + + :param key: base64 serialized public key + :type key: str + :param auth: authorization token + type key: str + + """ + tokens = auth.rsplit(' ', 1)[1].rsplit('.', 1) + kp = cls().from_raw_public(key.encode()) + return kp.verify_token( + validation_token=tokens[0].encode(), + verification_token=tokens[1] + ) + + @property + def private_key(self): + """The VAPID private ECDSA key""" + if not self._private_key: + raise VapidException("No private key. Call generate_keys()") + return self._private_key + + @private_key.setter + def private_key(self, value): + """Set the VAPID private ECDSA key + + :param value: the byte array containing the private ECDSA key data + :type value: ec.EllipticCurvePrivateKey + + """ + self._private_key = value + if value: + self._public_key = self.private_key.public_key() + + @property + def public_key(self): + """The VAPID public ECDSA key + + The public key is currently read only. Set it via the `.private_key` + method. This will autogenerate a public and private key if no value + has been set. + + :returns ec.EllipticCurvePublicKey + + """ + return self._public_key + + def generate_keys(self): + """Generate a valid ECDSA Key Pair.""" + self.private_key = ec.generate_private_key(ec.SECP256R1, + default_backend()) + + def private_pem(self): + return self.private_key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption() + ) + + def public_pem(self): + return self.public_key.public_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PublicFormat.SubjectPublicKeyInfo + ) + + def save_key(self, key_file): + """Save the private key to a PEM file. + + :param key_file: The file path to save the private key data + :type key_file: str + + """ + with open(key_file, "wb") as file: + file.write(self.private_pem()) + file.close() + + def save_public_key(self, key_file): + """Save the public key to a PEM file. + :param key_file: The name of the file to save the public key + :type key_file: str + + """ + with open(key_file, "wb") as file: + file.write(self.public_pem()) + file.close() + + def verify_token(self, validation_token, verification_token): + """Internally used to verify the verification token is correct. + + :param validation_token: Provided validation token string + :type validation_token: str + :param verification_token: Generated verification token + :type verification_token: str + :returns: Boolean indicating if verifictation token is valid. + :rtype: boolean + + """ + hsig = b64urldecode(verification_token.encode('utf8')) + r = int(binascii.hexlify(hsig[:32]), 16) + s = int(binascii.hexlify(hsig[32:]), 16) + try: + self.public_key.verify( + ecutils.encode_dss_signature(r, s), + validation_token, + signature_algorithm=ec.ECDSA(hashes.SHA256()) + ) + return True + except InvalidSignature: + return False + + def _base_sign(self, claims): + cclaims = copy.deepcopy(claims) + if not cclaims.get('exp'): + cclaims['exp'] = int(time.time()) + 86400 + if not self.conf.get('no-strict', False): + valid = _check_sub(cclaims.get('sub', '')) + else: + valid = cclaims.get('sub') is not None + if not valid: + raise VapidException( + "Missing 'sub' from claims. " + "'sub' is your admin email as a mailto: link.") + if not re.match(r"^https?://[^/:]+(:\d+)?$", + cclaims.get("aud", ""), + re.IGNORECASE): + raise VapidException( + "Missing 'aud' from claims. " + "'aud' is the scheme, host and optional port for this " + "transaction e.g. https://example.com:8080") + return cclaims + + def sign(self, claims, crypto_key=None): + """Sign a set of claims. + :param claims: JSON object containing the JWT claims to use. + :type claims: dict + :param crypto_key: Optional existing crypto_key header content. The + vapid public key will be appended to this data. + :type crypto_key: str + :returns: a hash containing the header fields to use in + the subscription update. + :rtype: dict + + """ + sig = sign(self._base_sign(claims), self.private_key) + pkey = 'p256ecdsa=' + pkey += b64urlencode( + self.public_key.public_bytes( + serialization.Encoding.X962, + serialization.PublicFormat.UncompressedPoint + )) + if crypto_key: + crypto_key = crypto_key + ';' + pkey + else: + crypto_key = pkey + + return {"Authorization": "{} {}".format(self._schema, sig.strip('=')), + "Crypto-Key": crypto_key} + + +class Vapid02(Vapid01): + """Minimal Vapid RFC8292 signature generation library + + https://tools.ietf.org/html/rfc8292 + + """ + _schema = "vapid" + + def sign(self, claims, crypto_key=None): + sig = sign(self._base_sign(claims), self.private_key) + pkey = self.public_key.public_bytes( + serialization.Encoding.X962, + serialization.PublicFormat.UncompressedPoint + ) + return{ + "Authorization": "{schema} t={t},k={k}".format( + schema=self._schema, + t=sig, + k=b64urlencode(pkey) + ) + } + + @classmethod + def verify(cls, auth): + pref_tok = auth.rsplit(' ', 1) + assert pref_tok[0].lower() == cls._schema, ( + "Incorrect schema specified") + parts = {} + for tok in pref_tok[1].split(','): + kv = tok.split('=', 1) + parts[kv[0]] = kv[1] + assert 'k' in parts.keys(), ( + "Auth missing public key 'k' value") + assert 't' in parts.keys(), ( + "Auth missing token set 't' value") + kp = cls().from_raw_public(parts['k'].encode()) + tokens = parts['t'].rsplit('.', 1) + return kp.verify_token( + validation_token=tokens[0].encode(), + verification_token=tokens[1] + ) + +def _check_sub(sub): + pattern =( + r"^(mailto:.+@((localhost|[%\w-]+(\.[%\w-]+)+|([0-9a-f]{1,4}):+([0-9a-f]{1,4})?)))|https:\/\/(localhost|[\w-]+\.[\w\.-]+|([0-9a-f]{1,4}:+)+([0-9a-f]{1,4})?)$" + ) + return re.match(pattern, sub, re.IGNORECASE) is not None + + +Vapid = Vapid02 diff --git a/dist/ba_data/python-site-packages/py_vapid/jwt.py b/dist/ba_data/python-site-packages/py_vapid/jwt.py new file mode 100644 index 0000000..c460f76 --- /dev/null +++ b/dist/ba_data/python-site-packages/py_vapid/jwt.py @@ -0,0 +1,87 @@ +import binascii +import json + +from cryptography.exceptions import InvalidSignature +from cryptography.hazmat.primitives.asymmetric import ec, utils +from cryptography.hazmat.primitives import hashes + +from py_vapid.utils import b64urldecode, b64urlencode, num_to_bytes + + +def extract_signature(auth): + """Extracts the payload and signature from a JWT, converting from RFC7518 + to RFC 3279 + + :param auth: A JWT Authorization Token. + :type auth: str + + :return tuple containing the signature material and signature + + """ + payload, asig = auth.encode('utf8').rsplit(b'.', 1) + sig = b64urldecode(asig) + if len(sig) != 64: + raise InvalidSignature() + + encoded = utils.encode_dss_signature( + s=int(binascii.hexlify(sig[32:]), 16), + r=int(binascii.hexlify(sig[:32]), 16) + ) + return payload, encoded + + +def decode(token, key): + """Decode a web token into an assertion dictionary + + :param token: VAPID auth token + :type token: str + :param key: bitarray containing the public key + :type key: str + + :return dict of the VAPID claims + + :raise InvalidSignature + + """ + try: + sig_material, signature = extract_signature(token) + dkey = b64urldecode(key.encode('utf8')) + pkey = ec.EllipticCurvePublicKey.from_encoded_point( + ec.SECP256R1(), + dkey, + ) + pkey.verify( + signature, + sig_material, + ec.ECDSA(hashes.SHA256()) + ) + return json.loads( + b64urldecode(sig_material.split(b'.')[1]).decode('utf8') + ) + except InvalidSignature: + raise + except(ValueError, TypeError, binascii.Error): + raise InvalidSignature() + + +def sign(claims, key): + """Sign the claims + + :param claims: list of JWS claims + :type claims: dict + :param key: Private key for signing + :type key: ec.EllipticCurvePrivateKey + :param algorithm: JWT "alg" descriptor + :type algorithm: str + + """ + header = b64urlencode(b"""{"typ":"JWT","alg":"ES256"}""") + # Unfortunately, chrome seems to require the claims to be sorted. + claims = b64urlencode(json.dumps(claims, + separators=(',', ':'), + sort_keys=True).encode('utf8')) + token = "{}.{}".format(header, claims) + rsig = key.sign(token.encode('utf8'), ec.ECDSA(hashes.SHA256())) + (r, s) = utils.decode_dss_signature(rsig) + sig = b64urlencode(num_to_bytes(r, 32) + num_to_bytes(s, 32)) + return "{}.{}".format(token, sig) diff --git a/dist/ba_data/python-site-packages/py_vapid/main.py b/dist/ba_data/python-site-packages/py_vapid/main.py new file mode 100644 index 0000000..58ca491 --- /dev/null +++ b/dist/ba_data/python-site-packages/py_vapid/main.py @@ -0,0 +1,115 @@ +# This Source Code Form is subject to the terms of the Mozilla Public +# License, v. 2.0. If a copy of the MPL was not distributed with this +# file, You can obtain one at http://mozilla.org/MPL/2.0/. + +import argparse +import os +import json + +from cryptography.hazmat.primitives import serialization + +from py_vapid import Vapid01, Vapid02, b64urlencode + + +def prompt(prompt): + # Not sure why, but python3 throws and exception if you try to + # monkeypatch for this. It's ugly, but this seems to play nicer. + try: + return input(prompt) + except NameError: + return raw_input(prompt) # noqa: F821 + + +def main(): + parser = argparse.ArgumentParser(description="VAPID tool") + parser.add_argument('--sign', '-s', help='claims file to sign') + parser.add_argument('--gen', '-g', help='generate new key pairs', + default=False, action="store_true") + parser.add_argument('--version2', '-2', help="use RFC8292 VAPID spec", + default=True, action="store_true") + parser.add_argument('--version1', '-1', help="use VAPID spec Draft-01", + default=False, action="store_true") + parser.add_argument('--json', help="dump as json", + default=False, action="store_true") + parser.add_argument('--no-strict', help='Do not be strict about "sub"', + default=False, action="store_true") + parser.add_argument('--applicationServerKey', + help="show applicationServerKey value", + default=False, action="store_true") + parser.add_argument('--private-key', '-k', help='private key pem file', + default="private_key.pem") + args = parser.parse_args() + + # Added to solve 2.7 => 3.* incompatibility + Vapid = Vapid02 + if args.version1: + Vapid = Vapid01 + if args.gen or not os.path.exists(args.private_key): + if not args.gen: + print("No private key file found.") + answer = None + while answer not in ['y', 'n']: + answer = prompt("Do you want me to create one for you? (Y/n)") + if not answer: + answer = 'y' + answer = answer.lower()[0] + if answer == 'n': + print("Sorry, can't do much for you then.") + exit(1) + vapid = Vapid(conf=args) + vapid.generate_keys() + print("Generating private_key.pem") + vapid.save_key('private_key.pem') + print("Generating public_key.pem") + vapid.save_public_key('public_key.pem') + vapid = Vapid.from_file(args.private_key) + claim_file = args.sign + result = dict() + if args.applicationServerKey: + raw_pub = vapid.public_key.public_bytes( + serialization.Encoding.X962, + serialization.PublicFormat.UncompressedPoint + ) + print("Application Server Key = {}\n\n".format( + b64urlencode(raw_pub))) + if claim_file: + if not os.path.exists(claim_file): + print("No {} file found.".format(claim_file)) + print(""" +The claims file should be a JSON formatted file that holds the +information that describes you. There are three elements in the claims +file you'll need: + + "sub" This is your site's admin email address + (e.g. "mailto:admin@example.com") + "exp" This is the expiration time for the claim in seconds. If you don't + have one, I'll add one that expires in 24 hours. + +You're also welcome to add additional fields to the claims which could be +helpful for the Push Service operations team to pass along to your operations +team (e.g. "ami-id": "e-123456", "cust-id": "a3sfa10987"). Remember to keep +these values short to prevent some servers from rejecting the transaction due +to overly large headers. See https://jwt.io/introduction/ for details. + +For example, a claims.json file could contain: + +{"sub": "mailto:admin@example.com"} +""") + exit(1) + try: + claims = json.loads(open(claim_file).read()) + result.update(vapid.sign(claims)) + except Exception as exc: + print("Crap, something went wrong: {}".format(repr(exc))) + raise exc + if args.json: + print(json.dumps(result)) + return + print("Include the following headers in your request:\n") + for key, value in result.items(): + print("{}: {}\n".format(key, value)) + print("\n") + + +if __name__ == '__main__': + main() diff --git a/dist/ba_data/python-site-packages/py_vapid/tests/test_vapid.py b/dist/ba_data/python-site-packages/py_vapid/tests/test_vapid.py new file mode 100644 index 0000000..ec11218 --- /dev/null +++ b/dist/ba_data/python-site-packages/py_vapid/tests/test_vapid.py @@ -0,0 +1,282 @@ +import binascii +import base64 +import copy +import os +import json +import unittest +from cryptography.hazmat.primitives import serialization +from mock import patch, Mock + +from py_vapid import Vapid01, Vapid02, VapidException, _check_sub +from py_vapid.jwt import decode + +TEST_KEY_PRIVATE_DER = """ +MHcCAQEEIPeN1iAipHbt8+/KZ2NIF8NeN24jqAmnMLFZEMocY8RboAoGCCqGSM49 +AwEHoUQDQgAEEJwJZq/GN8jJbo1GGpyU70hmP2hbWAUpQFKDByKB81yldJ9GTklB +M5xqEwuPM7VuQcyiLDhvovthPIXx+gsQRQ== +""" + +key = dict( + d=111971876876285331364078054667935803036831194031221090723024134705696601261147, # noqa + x=7512698603580564493364310058109115206932767156853859985379597995200661812060, # noqa + y=74837673548863147047276043384733294240255217876718360423043754089982135570501 # noqa +) + +# This is the same private key, in PEM form. +TEST_KEY_PRIVATE_PEM = ( + "-----BEGIN PRIVATE KEY-----{}" + "-----END PRIVATE KEY-----\n").format(TEST_KEY_PRIVATE_DER) + +# This is the same private key, as a point in uncompressed form. This should +# be Base64url-encoded without padding. +TEST_KEY_PRIVATE_RAW = """ +943WICKkdu3z78pnY0gXw143biOoCacwsVkQyhxjxFs +""".strip().encode('utf8') + +# This is a public key in PEM form. +TEST_KEY_PUBLIC_PEM = """-----BEGIN PUBLIC KEY----- +MFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAEEJwJZq/GN8jJbo1GGpyU70hmP2hb +WAUpQFKDByKB81yldJ9GTklBM5xqEwuPM7VuQcyiLDhvovthPIXx+gsQRQ== +-----END PUBLIC KEY----- +""" + +# this is a public key in uncompressed form ('\x04' + 2 * 32 octets) +# Remember, this should have any padding stripped. +TEST_KEY_PUBLIC_RAW = ( + "BBCcCWavxjfIyW6NRhqclO9IZj9oW1gFKUBSgwcigfNc" + "pXSfRk5JQTOcahMLjzO1bkHMoiw4b6L7YTyF8foLEEU" + ).strip('=').encode('utf8') + + +def setup_module(self): + with open('/tmp/private', 'w') as ff: + ff.write(TEST_KEY_PRIVATE_PEM) + with open('/tmp/public', 'w') as ff: + ff.write(TEST_KEY_PUBLIC_PEM) + with open('/tmp/private.der', 'w') as ff: + ff.write(TEST_KEY_PRIVATE_DER) + + +def teardown_module(self): + os.unlink('/tmp/private') + os.unlink('/tmp/public') + + +class VapidTestCase(unittest.TestCase): + def check_keys(self, v): + assert v.private_key.private_numbers().private_value == key.get('d') + assert v.public_key.public_numbers().x == key.get('x') + assert v.public_key.public_numbers().y == key.get('y') + + def test_init(self): + v1 = Vapid01.from_file("/tmp/private") + self.check_keys(v1) + v2 = Vapid01.from_pem(TEST_KEY_PRIVATE_PEM.encode()) + self.check_keys(v2) + v3 = Vapid01.from_der(TEST_KEY_PRIVATE_DER.encode()) + self.check_keys(v3) + v4 = Vapid01.from_file("/tmp/private.der") + self.check_keys(v4) + no_exist = '/tmp/not_exist' + Vapid01.from_file(no_exist) + assert os.path.isfile(no_exist) + os.unlink(no_exist) + + def repad(self, data): + return data + "===="[len(data) % 4:] + + @patch("py_vapid.Vapid01.from_pem", side_effect=Exception) + def test_init_bad_read(self, mm): + self.assertRaises(Exception, + Vapid01.from_file, + private_key_file="/tmp/private") + + def test_gen_key(self): + v = Vapid01() + v.generate_keys() + assert v.public_key + assert v.private_key + + def test_private_key(self): + v = Vapid01() + self.assertRaises(VapidException, + lambda: v.private_key) + + def test_public_key(self): + v = Vapid01() + assert v._private_key is None + assert v._public_key is None + + def test_save_key(self): + v = Vapid01() + v.generate_keys() + v.save_key("/tmp/p2") + os.unlink("/tmp/p2") + + def test_same_public_key(self): + v = Vapid01() + v.generate_keys() + v.save_public_key("/tmp/p2") + os.unlink("/tmp/p2") + + def test_from_raw(self): + v = Vapid01.from_raw(TEST_KEY_PRIVATE_RAW) + self.check_keys(v) + + def test_from_string(self): + v1 = Vapid01.from_string(TEST_KEY_PRIVATE_DER) + v2 = Vapid01.from_string(TEST_KEY_PRIVATE_RAW.decode()) + self.check_keys(v1) + self.check_keys(v2) + + def test_sign_01(self): + v = Vapid01.from_string(TEST_KEY_PRIVATE_DER) + claims = {"aud": "https://example.com", + "sub": "mailto:admin@example.com"} + result = v.sign(claims, "id=previous") + assert result['Crypto-Key'] == ( + 'id=previous;p256ecdsa=' + TEST_KEY_PUBLIC_RAW.decode('utf8')) + pkey = binascii.b2a_base64( + v.public_key.public_bytes( + serialization.Encoding.X962, + serialization.PublicFormat.UncompressedPoint + ) + ).decode('utf8').replace('+', '-').replace('/', '_').strip() + items = decode(result['Authorization'].split(' ')[1], pkey) + for k in claims: + assert items[k] == claims[k] + result = v.sign(claims) + assert result['Crypto-Key'] == ('p256ecdsa=' + + TEST_KEY_PUBLIC_RAW.decode('utf8')) + # Verify using the same function as Integration + # this should ensure that the r,s sign values are correctly formed + assert Vapid01.verify( + key=result['Crypto-Key'].split('=')[1], + auth=result['Authorization'] + ) + + def test_sign_02(self): + v = Vapid02.from_file("/tmp/private") + claims = {"aud": "https://example.com", + "sub": "mailto:admin@example.com", + "foo": "extra value"} + claim_check = copy.deepcopy(claims) + result = v.sign(claims, "id=previous") + auth = result['Authorization'] + assert auth[:6] == 'vapid ' + assert ' t=' in auth + assert ',k=' in auth + parts = auth[6:].split(',') + assert len(parts) == 2 + t_val = json.loads(base64.urlsafe_b64decode( + self.repad(parts[0][2:].split('.')[1]) + ).decode('utf8')) + k_val = binascii.a2b_base64(self.repad(parts[1][2:])) + assert binascii.hexlify(k_val)[:2] == b'04' + assert len(k_val) == 65 + assert claims == claim_check + for k in claims: + assert t_val[k] == claims[k] + + def test_sign_02_localhost(self): + v = Vapid02.from_file("/tmp/private") + claims = {"aud": "http://localhost:8000", + "sub": "mailto:admin@example.com", + "foo": "extra value"} + result = v.sign(claims, "id=previous") + auth = result['Authorization'] + assert auth[:6] == 'vapid ' + assert ' t=' in auth + assert ',k=' in auth + + def test_integration(self): + # These values were taken from a test page. DO NOT ALTER! + key = ("BDd3_hVL9fZi9Ybo2UUzA284WG5FZR30_95YeZJsiApwXKpNcF1rRPF3foI" + "iBHXRdJI2Qhumhf6_LFTeZaNndIo") + auth = ("eyJ0eXAiOiJKV1QiLCJhbGciOiJFUzI1NiJ9.eyJhdWQiOiJod" + "HRwczovL3VwZGF0ZXMucHVzaC5zZXJ2aWNlcy5tb3ppbGxhLmNvbSIsImV" + "4cCI6MTQ5NDY3MTQ3MCwic3ViIjoibWFpbHRvOnNpbXBsZS1wdXNoLWRlb" + "W9AZ2F1bnRmYWNlLmNvLnVrIn0.LqPi86T-HJ71TXHAYFptZEHD7Wlfjcc" + "4u5jYZ17WpqOlqDcW-5Wtx3x1OgYX19alhJ9oLumlS2VzEvNioZolQA") + assert Vapid01.verify(key=key, auth="webpush {}".format(auth)) + assert Vapid02.verify(auth="vapid t={},k={}".format(auth, key)) + + def test_bad_integration(self): + # These values were taken from a test page. DO NOT ALTER! + key = ("BDd3_hVL9fZi9Ybo2UUzA284WG5FZR30_95YeZJsiApwXKpNcF1rRPF3foI" + "iBHXRdJI2Qhumhf6_LFTeZaNndIo") + auth = ("WebPush eyJ0eXAiOiJKV1QiLCJhbGciOiJFUzI1NiJ9.eyJhdWQiOiJod" + "HRwczovL3VwZGF0ZXMucHVzaC5zZXJ2aWNlcy5tb3ppbGxhLmNvbSIsImV" + "4cCI6MTQ5NDY3MTQ3MCwic3ViIjoibWFpbHRvOnNpbXBsZS1wdXNoLWRlb" + "W9AZ2F1bnRmYWNlLmNvLnVrIn0.LqPi86T-HJ71TXHAYFptZEHD7Wlfjcc" + "4u5jYZ17WpqOlqDcW-5Wtx3x1OgYX19alhJ9oLumlS2VzEvNioZ_BAD") + assert Vapid01.verify(key=key, auth=auth) == False + + def test_bad_sign(self): + v = Vapid01.from_file("/tmp/private") + self.assertRaises(VapidException, + v.sign, + {}) + self.assertRaises(VapidException, + v.sign, + {'sub': 'foo', + 'aud': "p.example.com"}) + self.assertRaises(VapidException, + v.sign, + {'sub': 'mailto:foo@bar.com', + 'aud': "p.example.com"}) + self.assertRaises(VapidException, + v.sign, + {'sub': 'mailto:foo@bar.com', + 'aud': "https://p.example.com:8080/"}) + + def test_ignore_sub(self): + v = Vapid02.from_file("/tmp/private") + v.conf['no-strict'] = True + assert v.sign({"sub": "foo", "aud": "http://localhost:8000"}) + + @patch('cryptography.hazmat.primitives.asymmetric' + '.ec.EllipticCurvePublicNumbers') + def test_invalid_sig(self, mm): + from cryptography.exceptions import InvalidSignature + ve = Mock() + ve.verify.side_effect = InvalidSignature + pk = Mock() + pk.public_key.return_value = ve + mm.from_encoded_point.return_value = pk + self.assertRaises(InvalidSignature, + decode, + 'foo.bar.blat', + 'aaaa') + self.assertRaises(InvalidSignature, + decode, + 'foo.bar.a', + 'aaaa') + + def test_sub(self): + valid = [ + 'mailto:me@localhost', + 'mailto:me@1.2.3.4', + 'mailto:me@1234::', + 'mailto:me@1234::5678', + 'mailto:admin@example.org', + 'mailto:admin-test-case@example-test-case.test.org', + 'https://localhost', + 'https://exmample-test-case.test.org', + 'https://8001::', + 'https://8001:1000:0001', + 'https://1.2.3.4' + ] + invalid = [ + 'mailto:@foobar.com', + 'mailto:example.org', + 'mailto:0123:', + 'mailto:::1234', + 'https://somehost', + 'https://xyz:123', + ] + + for val in valid: + assert _check_sub(val) is True + for val in invalid: + assert _check_sub(val) is False diff --git a/dist/ba_data/python-site-packages/py_vapid/utils.py b/dist/ba_data/python-site-packages/py_vapid/utils.py new file mode 100644 index 0000000..02ecf4e --- /dev/null +++ b/dist/ba_data/python-site-packages/py_vapid/utils.py @@ -0,0 +1,39 @@ +import base64 +import binascii + + +def b64urldecode(data): + """Decodes an unpadded Base64url-encoded string. + + :param data: data bytes to decode + :type data: bytes + + :returns bytes + + """ + return base64.urlsafe_b64decode(data + b"===="[len(data) % 4:]) + + +def b64urlencode(data): + """Encode a byte string into a Base64url-encoded string without padding + + :param data: data bytes to encode + :type data: bytes + + :returns str + + """ + return base64.urlsafe_b64encode(data).replace(b'=', b'').decode('utf8') + + +def num_to_bytes(n, pad_to): + """Returns the byte representation of an integer, in big-endian order. + :param n: The integer to encode. + :type n: int + :param pad_to: Expected length of result, zeropad if necessary. + :type pad_to: int + :returns bytes + """ + h = '%x' % n + r = binascii.unhexlify('0' * (len(h) % 2) + h) + return b'\x00' * (pad_to - len(r)) + r