Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2025-06-25 08:26
📄 __init__.py	0 B	2020-09-21 22:03
📄 test_challenges.py # Copyright 2015 Google Inc. All rights reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Tests for the reauth module.""" import base64 import json import os import unittest import mock from google_reauth import challenges, errors import pyu2f class _U2FInterfaceMock(object): def Authenticate(self, unused_app_id, challenge, unused_registered_keys): raise self.error _u2f_interface_mock = _U2FInterfaceMock() class ChallengesTest(unittest.TestCase): """This class contains tests for reauth challanges. """ @mock.patch('pyu2f.u2f.GetLocalU2FInterface', return_value = _u2f_interface_mock) def testSecurityKeyError(self, u2f_mock): metadata = { 'status': 'READY', 'challengeId': 2, 'challengeType': 'SECURITY_KEY', 'securityKey': { 'applicationId': 'security_key_application_id', 'challenges': [{ 'keyHandle': 'some_key', 'challenge': base64.urlsafe_b64encode( 'some_challenge'.encode('ascii')).decode('ascii'), }] }} challenge = challenges.SecurityKeyChallenge() _u2f_interface_mock.error = pyu2f.errors.U2FError( pyu2f.errors.U2FError.DEVICE_INELIGIBLE) self.assertEqual(None, challenge.obtain_challenge_input(metadata)) _u2f_interface_mock.error = pyu2f.errors.U2FError( pyu2f.errors.U2FError.TIMEOUT) self.assertEqual(None, challenge.obtain_challenge_input(metadata)) _u2f_interface_mock.error = pyu2f.errors.NoDeviceFoundError() self.assertEqual(None, challenge.obtain_challenge_input(metadata)) _u2f_interface_mock.error = pyu2f.errors.U2FError( pyu2f.errors.U2FError.BAD_REQUEST) with self.assertRaises(pyu2f.errors.U2FError): challenge.obtain_challenge_input(metadata) _u2f_interface_mock.error = pyu2f.errors.UnsupportedVersionException() with self.assertRaises(pyu2f.errors.UnsupportedVersionException): challenge.obtain_challenge_input(metadata) @mock.patch('getpass.getpass', return_value = None) def testNoPassword(self, getpass_mock): self.assertEqual(challenges.PasswordChallenge().obtain_challenge_input({}), {'credential': ' '}) def testSaml(self): metadata = { 'status': 'READY', 'challengeId': 1, 'challengeType': 'SAML', 'securityKey': {} } challenge = challenges.SamlChallenge() self.assertEqual(True, challenge.is_locally_eligible) with self.assertRaises(errors.ReauthSamlLoginRequiredError): challenge.obtain_challenge_input(metadata)	3.2 KB	2020-09-21 22:03
📄 test_reauth.py # Copyright 2015 Google Inc. All rights reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Tests for the reauth module.""" import base64 import json import os import unittest import mock from six.moves import http_client from six.moves import urllib from oauth2client import client from google_reauth import challenges from google_reauth import reauth from google_reauth import errors from google_reauth import reauth_creds from google_reauth import _reauth_client from google_reauth.reauth_creds import Oauth2WithReauthCredentials from pyu2f import model from pyu2f import u2f _ok_response = mock.Mock(spec=['status'], status=http_client.OK) _error_response = mock.Mock(spec=['status'], status=None) class ReauthTest(unittest.TestCase): """This class contains tests for mocking reauth. The tests here are a bit more verbose since we are trying to mock out U2F, and present the reauth code some form of valid reply. This makes this test a bit of an integration test, as opposed to testing every individual method by itself. """ rapt_token = 'encoded_proof_of_reauth_token' correct_password = 'correct_password' oauth_api_url = 'http://some_url.com' client_id = 'some_id' client_secret = 'some_secret' def _request_mock_side_effect(self, args, kwargs): """Helper function to respond with valid requests as if a real server. This is the helper function for mocking HTTP calls. The calls that should end up here are to the oauth2 API or to the reauth API. The order of ifs tries to mimic the real order that the requests are expected, but we do not enforce a particular order so it can be more general. Args: args: Every arg passed to a request. *kwargs: Every keyed arg passed to a request. Returns: (str, str), Mocked (headers, content) Raises: Exception: In case this function doesn't know how to mock a request. """ # Requests to oauth2 have the body urlencoded. # Requests to reauth have a JSON body. # Try to decode both and use as needed. qp = dict(urllib.parse.parse_qsl(kwargs['body'])) try: qp_json = json.loads(kwargs['body']) except ValueError: qp_json = {} uri = kwargs['uri'] if 'uri' in kwargs else args[0] # First call to oauth2 has REAUTH_SCOPE and returns an access token. if ((uri == self.oauth_api_url and qp.get('scope') == reauth._REAUTH_SCOPE)): return _ok_response, json.dumps({'access_token': 'access_token_for_reauth'}) # Initialization call for reauth, serve first challenge if uri == (_reauth_client._REAUTH_API + ':start'): return None, json.dumps({ 'status': 'CHALLENGE_REQUIRED', 'sessionId': 'session_id_1', 'challenges': [{ 'status': 'READY', 'challengeId': 1, 'challengeType': 'PASSWORD', 'securityKey': {}, }], }) # Continuation call for reauth, check first challenge and serve the second if uri == (_reauth_client._REAUTH_API + '/session_id_1:continue'): self.assertEqual(1, qp_json.get('challengeId')) self.assertEqual('RESPOND', qp_json.get('action')) if (qp_json.get('proposalResponse', {}).get('credential') == self.correct_password): # We got a correct password, go to security key return None, json.dumps({ 'status': 'CHALLENGE_REQUIRED', 'sessionId': 'session_id_2', 'challenges': [{ 'status': 'READY', 'challengeId': 2, 'challengeType': 'SECURITY_KEY', 'securityKey': { 'applicationId': 'security_key_application_id', 'challenges': [{ 'keyHandle': 'some_key', 'challenge': base64.urlsafe_b64encode( 'some_challenge'.encode('ascii')).decode('ascii'), }], }, }], }) else: # We got an incorrect password, ask again. # Normally, the sessionID should be different, but for keeping this # function simple, we are going to reuse session_id_1 to come back to # this if block. return None, json.dumps({ 'status': 'CHALLENGE_PENDING', 'sessionId': 'session_id_1', 'challenges': [{ 'status': 'READY', 'challengeId': 1, 'challengeType': 'PASSWORD', 'securityKey': {}, }], }) # Continuation call for reauth, check second challenge and serve token if uri == (_reauth_client._REAUTH_API + '/session_id_2:continue'): self.assertEqual(2, qp_json.get('challengeId')) self.assertEqual('RESPOND', qp_json.get('action')) return None, json.dumps({ 'status': 'AUTHENTICATED', 'sessionId': 'session_id_3', 'encodedProofOfReauthToken': self.rapt_token, }) raise Exception( 'Unexpected call :/\nURL {0}\n{1}'.format(uri, kwargs['body'])) # This U2F mock is made by looking into the implementation of the class and # making the minimum requirement to actually answer a challenge. class _U2FInterfaceMock(object): def Authenticate(self, unused_app_id, challenge, unused_registered_keys): client_data = model.ClientData( model.ClientData.TYP_AUTHENTICATION, challenge, 'some_origin') return model.SignResponse('key_handle'.encode(), 'resp'.encode(), client_data) def _call_reauth(self, request_mock, scopes=None): if os.environ.get('SK_SIGNING_PLUGIN') is not None: raise unittest.SkipTest('unset SK_SIGNING_PLUGIN.') return reauth.get_rapt_token( request_mock, self.client_id, self.client_secret, 'some_refresh_token', self.oauth_api_url, scopes=scopes) def StartPatch(self, args, *kwargs): patcher = mock.patch(args, *kwargs) self.addCleanup(patcher.stop) return patcher.start() ####### # Helper functions and classes above. # Actual tests below. ####### def setUp(self): self.u2f_local_interface_mock = self.StartPatch( 'pyu2f.u2f.GetLocalU2FInterface') self.u2f_local_interface_mock.return_value = self._U2FInterfaceMock() self.getpass_mock = self.StartPatch('getpass.getpass') self.getpass_mock.return_value = self.correct_password self.is_interactive_mock = self.StartPatch('sys.stdin') self.is_interactive_mock.isatty = lambda: True def testPassAndGnubbyReauth(self): with mock.patch('httplib2.Http.request', side_effect = self._request_mock_side_effect) as request_mock: reauth_result = self._call_reauth(request_mock) self.assertEqual(self.rapt_token, reauth_result) self.assertEqual(4, request_mock.call_count) def testPassWithScopes(self): with mock.patch('httplib2.Http.request', side_effect = self._request_mock_side_effect) as request_mock: reauth_result = self._call_reauth(request_mock, [ 'https://www.googleapis.com/auth/scope1', 'https://www.googleapis.com/auth/scope2']) self.assertEqual(self.rapt_token, reauth_result) self.assertEqual(4, request_mock.call_count) def testIncorrectPassThenPassAndGnubbyReauth(self): with mock.patch('httplib2.Http.request', side_effect = self._request_mock_side_effect) as request_mock: self.getpass_mock.return_value = None self.getpass_mock.side_effect = ['bad_pass', self.correct_password] reauth_result = self._call_reauth(request_mock) self.assertEqual(self.rapt_token, reauth_result) self.assertEqual(5, request_mock.call_count) def testNonInteractiveError(self): with mock.patch('httplib2.Http.request', side_effect = self._request_mock_side_effect) as request_mock: self.is_interactive_mock.isatty = lambda: False with self.assertRaises(errors.ReauthUnattendedError): unused_reauth_result = self._call_reauth(request_mock) @mock.patch('google_reauth.challenges.AVAILABLE_CHALLENGES', {}) def testChallengeNotSupported(self): with mock.patch('httplib2.Http.request', side_effect = self._request_mock_side_effect) as request_mock: with self.assertRaises(errors.ReauthFailError): reauth_result = self._call_reauth(request_mock) self.assertEqual(self.rapt_token, reauth_result) self.assertEqual(4, request_mock.call_count) @mock.patch('google_reauth.challenges.PasswordChallenge.is_locally_eligible', False) def testChallengeNotEligible(self): with mock.patch('httplib2.Http.request', side_effect = self._request_mock_side_effect) as request_mock: with self.assertRaises(errors.ReauthFailError): reauth_result = self._call_reauth(request_mock) self.assertEqual(self.rapt_token, reauth_result) self.assertEqual(4, request_mock.call_count) def accessTokenRefreshError(self, response, content): def side_effect(args, *kwargs): uri = kwargs['uri'] if 'uri' in kwargs else args[0] if (uri == self.oauth_api_url): return response, content raise Exception( 'Unexpected call :/\nURL {0}\n{1}'.format(uri, kwargs['body'])) with mock.patch('httplib2.Http.request', side_effect = side_effect) as request_mock: with self.assertRaises(errors.HttpAccessTokenRefreshError): reauth.refresh_access_token( request_mock, self.client_id, self.client_secret, 'some_refresh_token', self.oauth_api_url) self.assertEqual(1, request_mock.call_count) def testAccessTokenRefreshError(self): self.accessTokenRefreshError(_ok_response, "foo") self.accessTokenRefreshError(_error_response, "foo") self.accessTokenRefreshError(_error_response, json.dumps({ 'error': 'non_reauth_error'})) def reauthAccessTokenError(self, response, content): def side_effect(args, *kwargs): qp = dict(urllib.parse.parse_qsl(kwargs['body'])) try: qp_json = json.loads(kwargs['body']) except ValueError: qp_json = {} uri = kwargs['uri'] if 'uri' in kwargs else args[0] if ((uri == self.oauth_api_url and qp.get('scope') == reauth._REAUTH_SCOPE)): return response, content raise Exception( 'Unexpected call :/\nURL {0}\n{1}'.format(uri, kwargs['body'])) with mock.patch('httplib2.Http.request', side_effect = side_effect) as request_mock: with self.assertRaises(errors.ReauthAccessTokenRefreshError): reauth.get_rapt_token( request_mock, self.client_id, self.client_secret, 'some_refresh_token', self.oauth_api_url) self.assertEqual(1, request_mock.call_count) def testReauthAccessTokenError(self): self.reauthAccessTokenError(_ok_response, "foo") self.reauthAccessTokenError(_ok_response, json.dumps({})) self.reauthAccessTokenError( _error_response, json.dumps({'error': 'some error'})) def testResponseErrorMessage(self): self.assertEqual('Invalid response.', reauth._get_refresh_error_message({})) self.assertEqual('There was an error.', reauth._get_refresh_error_message({'error': 'There was an error.'})) self.assertEqual('There was an error: error description', reauth._get_refresh_error_message({ 'error': 'There was an error', 'error_description': 'error description'})) def getChallengesError(self, content): def side_effect(args, *kwargs): uri = kwargs['uri'] if 'uri' in kwargs else args[0] # First call to oauth2 has REAUTH_SCOPE and returns an access token. if ((uri == self.oauth_api_url)): return _ok_response, json.dumps({ 'access_token': 'access_token_for_reauth'}) # Initialization call for reauth, serve first challenge if uri == (_reauth_client._REAUTH_API + ':start'): return None, content with mock.patch('httplib2.Http.request', side_effect = side_effect) as request_mock: with self.assertRaises(errors.ReauthAPIError) as e: reauth.get_rapt_token( request_mock, self.client_id, self.client_secret, 'some_refresh_token', self.oauth_api_url) self.assertEqual(2, request_mock.call_count) def testGetChallengesError(self): self.getChallengesError(json.dumps({'status': 'ERROR'})) self.getChallengesError(json.dumps({ 'error': {'message': 'Some error'}})) def testChallangeNotActivatedError(self): def side_effect(args, **kwargs): uri = kwargs['uri'] if 'uri' in kwargs else args[0] # First call to oauth2 has REAUTH_SCOPE and returns an access token. if ((uri == self.oauth_api_url)): return _ok_response, json.dumps({ 'access_token': 'access_token_for_reauth'}) # Initialization call for reauth, serve first challenge if uri == (_reauth_client._REAUTH_API + ':start'): return None, json.dumps({ 'status': 'CHALLENGE_REQUIRED', 'sessionId': 'session_id_1', 'challenges': [{ 'status': 'NOT_READY', 'challengeId': 1, 'challengeType': 'PASSWORD'}]}) request_mock = self.StartPatch('httplib2.Http.request') request_mock.side_effect = side_effect with self.assertRaises(errors.ReauthFailError) as e: reauth.get_rapt_token( request_mock, self.client_id, self.client_secret, 'some_refresh_token', self.oauth_api_url, scopes=None) self.assertEqual(6, request_mock.call_count) @mock.patch('google_reauth.challenges.SecurityKeyChallenge.obtain_challenge_input', return_value = None) def testRetryOnNoUserInput(self, challenge_mock): with mock.patch('httplib2.Http.request', side_effect = self._request_mock_side_effect) as request_mock: with self.assertRaises(errors.ReauthFailError): reauth_result = self._call_reauth(request_mock) self.assertEqual(7, request_mock.call_count)	16.09 KB	2020-09-21 22:03
📄 test_reauth_creds.py # Copyright 2015 Google Inc. All rights reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Tests for the reauth module.""" import base64 import datetime import json import os import unittest import mock from mock import patch from six.moves import http_client from six.moves import urllib from oauth2client import client from oauth2client import client from google_reauth import reauth from google_reauth import errors from google_reauth import reauth_creds from google_reauth import _reauth_client from google_reauth.reauth_creds import Oauth2WithReauthCredentials _ok_response = lambda: None setattr(_ok_response, 'status', http_client.OK) _error_response = lambda: None setattr(_error_response, 'status', None) _token_response = ( _ok_response, json.dumps( {'access_token': 'new_access_token', 'refresh_token': 'new_refresh_token', 'expires_in': 60 * 60})) class MockStore(client.Storage): def __init__(self): self.credentials = None def locked_put(self, credentials): self.credentials = credentials def locked_get(self): return self.credentials class ReauthCredsTest(unittest.TestCase): """This class contains tests for Oauth2WithReauthCredentials. """ def StartPatch(self, args, kwargs): patcher = mock.patch(args, *kwargs) self.addCleanup(patcher.stop) return patcher.start() @patch('google_reauth.reauth.get_rapt_token') def get_rapt_token(self, args, *kwargs): return 'rapt_token' def _http_mock(self, request_side_effect): request_mock = self.StartPatch('httplib2.Http.request') request_mock.side_effect = request_side_effect http_mock = self.StartPatch('httplib2.Http') http_mock.request = request_mock return http_mock def _run_refresh_test(self, http_mock, access_token, refresh_token, token_expiry, invalid): creds = self._get_creds() store = MockStore() creds.set_store(store) creds._do_refresh_request(http_mock) self._check_credentials(creds, store, access_token, refresh_token, token_expiry, invalid) def _get_creds(self): return Oauth2WithReauthCredentials( access_token='old_token', client_id='id', client_secret='secret', refresh_token='old_refresh_token', token_expiry=datetime.datetime(2018, 3, 2, 21, 26, 13), token_uri='token_uri', user_agent='user_agent') def _check_credentials(self, creds, store, access_token, refresh_token, token_expiry, invalid): stored_creds = store.locked_get() if store else creds self.assertEqual(access_token, creds.access_token) self.assertEqual(access_token, stored_creds.access_token) self.assertEqual(refresh_token, creds.refresh_token) self.assertEqual(refresh_token, stored_creds.refresh_token) self.assertEqual(token_expiry, creds.token_expiry) self.assertEqual(token_expiry, stored_creds.token_expiry) self.assertEqual(invalid, creds.invalid) self.assertEqual(invalid, stored_creds.invalid) ####### # Helper functions and classes above. # Actual tests below. ####### def setUp(self): get_rapt = self.StartPatch('google_reauth.reauth.get_rapt_token') get_rapt.return_value='rapt_token' current_datetime = self.StartPatch('oauth2client.client._UTCNOW') current_datetime.return_value = datetime.datetime(2018, 3, 2, 21, 26, 13) def testFromOAuth2Credentials(self): orig = client.OAuth2Credentials( access_token='at', client_id='ci', client_secret='cs', refresh_token='rt', token_expiry='te', token_uri='tu', user_agent='ua') cred = Oauth2WithReauthCredentials.from_OAuth2Credentials(orig) self.assertEqual('Oauth2WithReauthCredentials', cred.__class__.__name__) self.assertEqual('ci', cred.client_id) self.assertEqual('cs', cred.client_secret) def testTokenExpiryFromJson(self): cred = Oauth2WithReauthCredentials.from_json(json.dumps({ 'access_token': 'access_token', 'client_id': 'client_id', 'client_secret': 'client_secret', 'refresh_token': 'refresh_token', 'token_expiry': 'foo', 'token_uri': 'token_uri', 'user_agent': 'user_agent', 'invalid': False})) self.assertEqual(None, cred.token_expiry) cred = Oauth2WithReauthCredentials.from_json(json.dumps({ 'access_token': 'access_token', 'client_id': 'client_id', 'client_secret': 'client_secret', 'refresh_token': 'refresh_token', 'token_expiry': '2018-03-02T21:26:13Z', 'token_uri': 'token_uri', 'user_agent': 'user_agent', 'invalid': False})) self.assertEqual(datetime.datetime(2018, 3, 2, 21, 26, 13), cred.token_expiry) def testRefreshNoReauthRequired(self): def request_side_effect(self, args, *kwargs): return _token_response self._run_refresh_test( self._http_mock(request_side_effect), 'new_access_token', 'new_refresh_token', datetime.datetime(2018, 3, 2, 22, 26, 13), False) def testRefreshReauthRequired(self): responses = [ _token_response, (_error_response, json.dumps({ 'error': 'invalid_grant', 'error_subtype': 'rapt_required'}))] def request_side_effect(self, args, *kwargs): return responses.pop() self._run_refresh_test( self._http_mock(request_side_effect), 'new_access_token', 'new_refresh_token', datetime.datetime(2018, 3, 2, 22, 26, 13), False) def testInvalidRapt(self): responses = [ (_error_response, json.dumps({ 'error': 'invalid_grant', 'error_subtype': 'rapt_required'})), (_error_response, json.dumps({ 'error': 'invalid_grant', 'error_subtype': 'rapt_required'}))] def request_side_effect(self, args, *kwargs): return responses.pop() creds = self._get_creds() store = MockStore() creds.set_store(store) with self.assertRaises(client.HttpAccessTokenRefreshError): creds._do_refresh_request(self._http_mock(request_side_effect)) self._check_credentials( creds, store, 'old_token', 'old_refresh_token', datetime.datetime(2018, 3, 2, 21, 26, 13), True) def testRefreshWithStore(self): def request_side_effect(self, args, *kwargs): return _token_response creds = self._get_creds() store = MockStore() creds.set_store(store) creds._do_refresh_request(self._http_mock(request_side_effect)) self._check_credentials( creds, store, 'new_access_token', 'new_refresh_token', datetime.datetime(2018, 3, 2, 22, 26, 13), False) def testRefreshNoStore(self): def request_side_effect(self, args, *kwargs): return _token_response creds = self._get_creds() creds._do_refresh_request(self._http_mock(request_side_effect)) self._check_credentials( creds, None, 'new_access_token', 'new_refresh_token', datetime.datetime(2018, 3, 2, 22, 26, 13), False) def testRefreshNoExpiry(self): def request_side_effect(self, args, **kwargs): return ( _ok_response, json.dumps( {'access_token': 'new_access_token', 'refresh_token': 'new_refresh_token'})) self._run_refresh_test( self._http_mock(request_side_effect), 'new_access_token', 'new_refresh_token', None, False)	8.53 KB	2020-09-21 22:03

Terminal