8000 ValidationClient for client validation, add ClientValidationJwt by codejudas · Pull Request #309 · twilio/twilio-python · GitHub
[go: up one dir, main page]
Skip to content

ValidationClient for client validation, add ClientValidationJwt #309

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Mar 2, 2017
Merged

ValidationClient for client validation, add ClientValidationJwt #309

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
77cb421
ValidationClient for client validation, add ClientValidationJwt
codejudas Feb 22, 2017
fed7324
Fixes after manual testing, add tests to validation client
codejudas Feb 24, 2017
f26d490
PR Comments
codejudas Mar 2, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions tests/requirements.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,3 +6,4 @@ nosexcover
flake8
mccabe
wheel>=0.22.0
cryptography
Empty file added tests/unit/http/__init__.py
View file
Open in desktop
Empty file.
107 changes: 107 additions & 0 deletions tests/unit/http/test_validation_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,107 @@
import unittest

import mock
from mock import patch, Mock
from requests import Request
from requests import Session

from twilio.http.validation_client import ValidationClient


class TestValidationClientHelpers(unittest.TestCase):
def setUp(self):
self.request = Request(
'GET',
'https://api.twilio.com/2010-04-01/Accounts/AC123/Messages',
auth=('Username', 'Password'),
)
self.request = self.request.prepare()
self.client = ValidationClient('AC123', 'SK123', 'CR123', 'private_key')

def test_build_validation_payload_basic(self):
validation_payload = self.client._build_validation_payload(self.request)
self.assertEqual('GET', validation_payload.method)
self.assertEqual('/2010-04-01/Accounts/AC123/Messages', validation_payload.path)
self.assertEqual('', validation_payload.query_string)
self.assertEqual(['authorization', 'host'], validation_payload.signed_headers)
self.assertEqual('', validation_payload.body)

def test_build_validation_payload_query_string_parsed(self):
self.request.url = self.request.url + '?QueryParam=1&Other=true'

validation_payload = self.client._build_validation_payload(self.request)

self.assertEqual('GET', validation_payload.method)
self.assertEqual('/2010-04-01/Accounts/AC123/Messages', validation_payload.path)
self.assertEqual('QueryParam=1&Other=true', validation_payload.query_string)
self.assertEqual(['authorization', 'host'], validation_payload.signed_headers)
self.assertEqual('', validation_payload.body)

def test_build_validation_payload_body_parsed(self):
self.request.body = 'foobar'

validation_payload = self.client._build_validation_payload(self.request)

self.assertEqual('GET', validation_payload.method)
self.assertEqual('/2010-04-01/Accounts/AC123/Messages', validation_payload.path)
self.assertEqual('', validation_payload.query_string)
self.assertEqual(['authorization', 'host'], validation_payload.signed_headers)
self.assertEqual('foobar', validation_payload.body)

def test_build_validation_payload_complex(self):
self.request.body = 'foobar'
self.request.url = self.request.url + '?QueryParam=Value&OtherQueryParam=OtherValue'

validation_payload = self.client._build_validation_payload(self.request)

self.assertEqual('GET', validation_payload.method)
self.assertEqual('/2010-04-01/Accounts/AC123/Messages', validation_payload.path)
self.assertEqual(['authorization', 'host'], validation_payload.signed_headers)
self.assertEqual('foobar', validation_payload.body)
self.assertEqual('QueryParam=Value&OtherQueryParam=OtherValue',
validation_payload.query_string)

def test_get_host(self):
self.assertEqual('api.twilio.com', self.client._get_host(self.request))


class TestValidationClientRequest(unittest.TestCase):

def setUp(self):
self.session_patcher = patch('twilio.http.validation_client.Session')
self.jwt_patcher = patch('twilio.http.validation_client.ClientValidationJwt')

self.session_mock = Mock(wraps=Session())
self.validation_token = self.jwt_patcher.start()
self.request_mock = Mock()

self.session_mock.prepare_request.return_value = self.request_mock
self.session_mock.send.return_value = Mock(status_code=200, content='test')
self.validation_token.return_value.to_jwt.return_value = 'test-token'
self.request_mock.headers = {}

session_constructor_mock = self.session_patcher.start()
session_constructor_mock.return_value = self.session_mock

self.client = ValidationClient('AC123', 'SK123', 'CR123', 'private_key')

def tearDown(self):
self.session_patcher.stop()
self.jwt_patcher.stop()

def test_request_does_not_overwrite_host_header(self):
self.request_mock.url = 'https://api.twilio.com/'

self.client.request('doesnt matter', 'doesnt matter')

self.assertEqual('api.twilio.com', self.request_mock.headers['Host'])
self.assertEqual('test-token', self.request_mock.headers['Twilio-Client-Validation'])

def test_request_sets_host_header_if_missing(self):
self.request_mock.url = 'https://api.twilio.com/'
self.request_mock.headers = {'Host': 'other.twilio.com'}

self.client.request('doesnt matter', 'doesnt matter')

self.assertEqual('other.twilio.com', self.request_mock.headers['Host'])
self.assertEqual('test-token', self.request_mock.headers['Twilio-Client-Validation'])
286 changes: 286 additions & 0 deletions tests/unit/jwt/test_client_validation.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,286 @@
import unittest
import time

from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives.serialization import (
Encoding,
PublicFormat,
PrivateFormat,
NoEncryption
)

from twilio.http.validation_client import ValidationPayload
from twilio.jwt import Jwt
from twilio.jwt.validation import ClientValidationJwt


class ClientValidationJwtTest(unittest.TestCase):
def test_generate_payload_basic(self):
vp = ValidationPayload(
method='GET',
path='https://api.twilio.com/',
query_string='q1=v1',
signed_headers=['headerb', 'headera'],
all_headers={'head': 'toe', 'headera': 'vala', 'headerb': 'valb'},
body='me=letop&you=leworst'
)

expected_payload = '\n'.join([
'GET',
'https://api.twilio.com/',
'q1=v1',
'headera:vala',
'headerb:valb',
'',
'headera;headerb',
'{}'.format(ClientValidationJwt._hash('me=letop&you=leworst'))
])
expected_payload = ClientValidationJwt._hash(expected_payload)

jwt = ClientValidationJwt('AC123', 'SK123', 'CR123', 'secret', vp)

actual_payload = jwt._generate_payload()
self.assertEqual('headera;headerb', actual_payload['hrh'])
self.assertEqual(expected_payload, actual_payload['rqh'])

def test_generate_payload_complex(self):
vp = ValidationPayload(
method='GET',
path='https://api.twilio.com/',
query_string='q1=v1&q2=v2&a=b',
signed_headers=['headerb', 'headera'],
all_headers={'head': 'toe', 'Headerb': 'valb', 'yeezy': 'weezy'},
body='me=letop&you=leworst'
)

expected_payload = '\n'.join([
'GET',
'https://api.twilio.com/',
'a=b&q1=v1&q2=v2',
'headerb:valb',
'',
'headera;headerb',
'{}'.format(ClientValidationJwt._hash('me=letop&you=leworst'))
])
expected_payload = ClientValidationJwt._hash(expected_payload)

jwt = ClientValidationJwt('AC123', 'SK123', 'CR123', 'secret', vp)

actual_payload = jwt._generate_payload()
self.assertEqual('headera;headerb', actual_payl 10000 oad['hrh'])
self.assertEqual(expected_payload, actual_payload['rqh'])

def test_generate_payload_no_query_string(self):
vp = ValidationPayload(
method='GET',
path='https://api.twilio.com/',
query_string='',
signed_headers=['headerb', 'headera'],
all_headers={'head': 'toe', 'Headerb': 'valb', 'yeezy': 'weezy'},
body='me=letop&you=leworst'
)

expected_payload = '\n'.join([
'GET',
'https://api.twilio.com/',
'',
'headerb:valb',
'',
'headera;headerb',
'{}'.format(ClientValidationJwt._hash('me=letop&you=leworst'))
])
expected_payload = ClientValidationJwt._hash(expected_payload)

jwt = ClientValidationJwt('AC123', 'SK123', 'CR123', 'secret', vp)

actual_payload = jwt._generate_payload()
self.assertEqual('headera;headerb', actual_payload['hrh'])
self.assertEqual(expected_payload, actual_payload['rqh'])

def test_generate_payload_no_req_body(self):
vp = ValidationPayload(
method='GET',
path='https://api.twilio.com/',
query_string='q1=v1',
signed_headers=['headerb', 'headera'],
all_headers={'head': 'toe', 'headera': 'vala', 'headerb': 'valb'},
body=''
)

expected_payload = '\n'.join([
'GET',
'https://api.twilio.com/',
'q1=v1',
'headera:vala',
'headerb:valb',
'',
'headera;headerb',
''
])
expected_payload = ClientValidationJwt._hash(expected_payload)

jwt = ClientValidationJwt('AC123', 'SK123', 'CR123', 'secret', vp)

actual_payload = jwt._generate_payload()
self.assertEqual('headera;headerb', actual_payload['hrh'])
self.assertEqual(expected_payload, actual_payload['rqh'])

def test_generate_payload_header_keys_lowercased(self):
vp = ValidationPayload(
method='GET',
path='https://api.twilio.com/',
query_string='q1=v1',
signed_headers=['headerb', 'headera'],
all_headers={'head': 'toe', 'Headera': 'vala', 'Headerb': 'valb'},
body='me=letop&you=leworst'
)

expected_payload = '\n'.join([
'GET',
'https://api.twilio.com/',
'q1=v1',
'headera:vala',
'headerb:valb',
'',
'headera;headerb',
'{}'.format(ClientValidationJwt._hash('me=letop&you=leworst'))
])
expected_payload = ClientValidationJwt._hash(expected_payload)

jwt = ClientValidationJwt('AC123', 'SK123', 'CR123', 'secret', vp)

actual_payload = jwt._generate_payload()
self.assertEqual('headera;headerb', actual_payload['hrh'])
self.assertEqual(expected_payload, actual_payload['rqh'])

def test_generate_payload_no_headers(self):
vp = ValidationPayload(
method='GET',
path='https://api.twilio.com/',
query_string='q1=v1',
signed_headers=['headerb', 'headera'],
all_headers={},
body='me=letop&you=leworst'
)

expected_payload = '\n'.join([
'GET',
'https://api.twilio.com/',
'q1=v1',
'',
'headera;headerb',
'{}'.format(ClientValidationJwt._hash('me=letop&you=leworst'))
])
expected_payload = ClientValidationJwt._hash(expected_payload)

jwt = ClientValidationJwt('AC123', 'SK123', 'CR123', 'secret', vp)

actual_payload = jwt._generate_payload()
self.assertEqual('headera;headerb', actual_payload['hrh'])
self.assertEqual(expected_payload, actual_payload['rqh'])

def test_generate_payload_schema_correct_1(self):
"""Test against a known good rqh payload hash"""
vp = ValidationPayload(
method='GET',
path='/Messages',
query_string='PageSize=5&Limit=10',
signed_headers=['authorization', 'host'],
all_headers={'authorization': 'foobar', 'host': 'api.twilio.com'},
body='foobar'
)

expected_hash = '4dc9b67bed579647914587b0e22a1c65c1641d8674797cd82de65e766cce5f80'

jwt = ClientValidationJwt('AC123', 'SK123', 'CR123', 'secret', vp)

actual_payload = jwt._generate_payload()
self.assertEqual('authorization;host', actual_payload['hrh'])
self.assertEqual(expected_hash, actual_payload['rqh'])

def test_generate_payload_schema_correct_2(self):
"""Test against a known good rqh payload hash"""
vp = ValidationPayload(
method='POST',
path='/Messages',
query_string='',
signed_headers=['authorization', 'host'],
all_headers={'authorization': 'foobar', 'host': 'api.twilio.com'},
body='testbody'
)

expected_hash = 'bd792c967c20d546c738b94068f5f72758a10d26c12979677501e1eefe58c65a'

jwt = ClientValidationJwt('AC123', 'SK123', 'CR123', 'secret', vp)

actual_payload = jwt._generate_payload()
self.assertEqual('authorization;host', actual_payload['hrh'])
self.assertEqual(expected_hash, actual_payload['rqh'])

def test_jwt_payload(self):
vp = ValidationPayload(
method='GET',
path='/Messages',
query_string='PageSize=5&Limit=10',
signed_headers=['authorization', 'host'],
all_headers={'authorization': 'foobar', 'host': 'api.twilio.com'},
body='foobar'
)
expected_hash = '4dc9b67bed579647914587b0e22a1c65c1641d8674797cd82de65e766cce5f80'

jwt = ClientValidationJwt('AC123', 'SK123', 'CR123', 'secret', vp)

self.assertDictContainsSubset({
'hrh': 'authorization;host',
'rqh': expected_hash,
'iss': 'SK123',
'sub': 'AC123',
}, jwt.payload)
self.assertGreaterEqual(jwt.payload['exp'], time.time(), 'JWT exp is before now')
self.assertLessEqual(jwt.payload['exp'], time.time() + 301, 'JWT exp is after now + 5mins')
self.assertDictEqual({
'alg': 'RS256',
'typ': 'JWT',
'cty': 'twilio-pkrv;v=1',
'kid': 'CR123'
}, jwt.headers)

def test_jwt_signing(self):
vp = ValidationPayload(
method='GET',
path='/Messages',
query_string='PageSize=5&Limit=10',
signed_headers=['authorization', 'host'],
all_headers={'authorization': 'foobar', 'host': 'api.twilio.com'},
body='foobar'
)
expected_hash = '4dc9b67bed579647914587b0e22a1c65c1641d8674797cd82de65e766cce5f80'

private_key = rsa.generate_private_key(
public_exponent=65537,
key_size=2048,
backend=default_backend()
)
public_key = private_key.public_key().public_bytes(Encoding.PEM, PublicFormat.PKCS1)
private_key = private_key.private_bytes(Encoding.PEM, PrivateFormat.PKCS8, NoEncryption())

jwt = ClientValidationJwt('AC123', 'SK123', 'CR123', private_key, vp)
decoded = Jwt.from_jwt(jwt.to_jwt(), public_key)

self.assertDictContainsSubset({
'hrh': 'authorization;host',
'rqh': expected_hash,
'iss': 'SK123',
'sub': 'AC123',
}, decoded.payload)
self.assertGreaterEqual(decoded.payload['exp'], time.time(), 'JWT exp is before now')
self.assertLessEqual(decoded.payload['exp'], time.time() + 501, 'JWT exp is after now + 5m')
self.assertDictEqual({
'alg': 'RS256',
'typ': 'JWT',
'cty': 'twilio-pkrv;v=1',
'kid': 'CR123'
}, decoded.headers)


Loading
0