work in progress: dummy oidc data handler for testing
This commit is contained in:
		
							parent
							
								
									f21910e675
								
							
						
					
					
						commit
						b2d1c7888b
					
				
					 4 changed files with 57 additions and 4 deletions
				
			
		|  | @ -2,6 +2,7 @@ | ||||||
| 
 | 
 | ||||||
| import logging | import logging | ||||||
| from os import getenv | from os import getenv | ||||||
|  | import sys | ||||||
| 
 | 
 | ||||||
| #from scopes.web.app import demo_app, zope_app | #from scopes.web.app import demo_app, zope_app | ||||||
| 
 | 
 | ||||||
|  | @ -32,8 +33,14 @@ dbuser = None | ||||||
| dbpassword = None | dbpassword = None | ||||||
| dbschema = None | dbschema = None | ||||||
| 
 | 
 | ||||||
|  | # special testing stuff | ||||||
|  | #from scopes.tests import oidc_data # add oidc URIs and keys to dummy_requests data | ||||||
|  | from scopes.tests import dummy_requests | ||||||
|  | sys.modules['requests'] = dummy_requests | ||||||
|  | 
 | ||||||
|  | 
 | ||||||
| # authentication settings | # authentication settings | ||||||
| oidc_provider = 'test://testing' | oidc_provider = 'test://oidc' | ||||||
| oidc_client_id = getenv('OIDC_CLIENT_ID', '12345') | oidc_client_id = getenv('OIDC_CLIENT_ID', '12345') | ||||||
| oidc_params = dict( | oidc_params = dict( | ||||||
|     op_config_url=oidc_provider + '/.well-known/openid-configuration', |     op_config_url=oidc_provider + '/.well-known/openid-configuration', | ||||||
|  |  | ||||||
							
								
								
									
										46
									
								
								scopes/tests/dummy_requests.py
									
										
									
									
									
										Normal file
									
								
							
							
						
						
									
										46
									
								
								scopes/tests/dummy_requests.py
									
										
									
									
									
										Normal file
									
								
							|  | @ -0,0 +1,46 @@ | ||||||
|  | # scopes.tests.requests | ||||||
|  | 
 | ||||||
|  | """Dummy requests implementation for testing.""" | ||||||
|  | 
 | ||||||
|  | from logging import getLogger | ||||||
|  | logger = getLogger('tests.dummy_requests') | ||||||
|  | 
 | ||||||
|  | def get(url, *args, **kw): | ||||||
|  |     logger.info(f'get: %s - %s - %s', url, args, kw) | ||||||
|  |     return FakeResponse(response_data[url]) | ||||||
|  | 
 | ||||||
|  | 
 | ||||||
|  | class FakeResponse: | ||||||
|  | 
 | ||||||
|  |     def __init__(self, data): | ||||||
|  |         self.data = data | ||||||
|  | 
 | ||||||
|  |     def json(self): | ||||||
|  |         return self.data | ||||||
|  | 
 | ||||||
|  | 
 | ||||||
|  | response_data = { | ||||||
|  |     'test://oidc/.well-known/openid-configuration': { | ||||||
|  |         "issuer": "test://oidc", | ||||||
|  |         "authorization_endpoint": "test://oidc/oauth/v2/authorize", | ||||||
|  |         "token_endpoint": "test://oidc/oauth/v2/token", | ||||||
|  |         "introspection_endpoint": "test://oidc/oauth/v2/introspect", | ||||||
|  |         "userinfo_endpoint": "test://oidc/oidc/v1/userinfo", | ||||||
|  |         "revocation_endpoint": "test://oidc/oauth/v2/revoke", | ||||||
|  |         "end_session_endpoint": "test://oidc/oidc/v1/end_session", | ||||||
|  |         "device_authorization_endpoint": "test://oidc/oauth/v2/device_authorization", | ||||||
|  |         "jwks_uri": "test://oidc/oauth/v2/keys"}, | ||||||
|  |     'test://oidc/oauth/v2/keys': { "keys": [ | ||||||
|  |        {"use": "sig", | ||||||
|  |         "kty": "RSA", | ||||||
|  |         "kid": "316638486247563085", | ||||||
|  |         "alg": "RS256", | ||||||
|  |         "n": "167qFCfRa0tRR0MZv-PQVwdiVFf0NtfN-zFAogRASm6437sbXfsfxkpbh1F77TwQdl4qlR5Na_Ecs8VTxOuyHmuhIJ4FyZV4M0h71KRw7LCTVuNw7mWLpbjKPBzidyhctbkJrkcKtJymnHELsct0CdT16Lb27phd_0cBJexGbwhVNQBs10VbkvUJHHOJe6A_JVS9Q3_3MEWyCyFoHPeMchlk_Gd6yMiH4aJ1ql3GZD6c2JB9crloTH_oPWWFQObGoXTKcFonEBdkrwuCQfRVOfGh8UIhIcTM0JNgqtQOCcIkf0emfI30SoWSc6Qz8lU70Vpmb3qQgsqATFICgzgABw", | ||||||
|  |         "e": "AQAB"}, | ||||||
|  |        {"use": "sig", | ||||||
|  |         "kty": "RSA", | ||||||
|  |         "kid": "316766976250797901", | ||||||
|  |         "alg": "RS256", | ||||||
|  |         "n": "yZKIsrUWT2fEj4OtUUFYQbEe_Clodz464tn5vMAQ0q8zV07bqFaA7WKuBflowYctDNxoxdbiFNISpKEOx6yFnx7_g6Zd46DWsj5ggGZvNkgOa9SqTIsA7ho9nk7LDLQRpV0k5N1HkiG66GUqUCV2llJhstpTDQQLDvhI3qussG2HyylpTQSu-9b6gry0rb397yjAnXQu6tFOubEDteTN0fLNMblcdd2AvZKpGA2o_-M5U6AckezfmBCBdHWmrwxpjGGf7KWqGg8j6bJkV3sMg4XfD2x0KNog_3D-0pSx6k8dSWZGkNlDxB5AdWvNDYg1stkvjeNEbIJAhv0-awLs9Q", | ||||||
|  |         "e": "AQAB"}]} | ||||||
|  | } | ||||||
|  | @ -24,6 +24,7 @@ def test_app(self, config): | ||||||
|     logger.info('test_app') |     logger.info('test_app') | ||||||
|     storage = config.storageFactory(config.dbschema) |     storage = config.storageFactory(config.dbschema) | ||||||
|     response = publishRequest(config, storage, '/top') |     response = publishRequest(config, storage, '/top') | ||||||
|  |     print('***', response.getStatus(), response.getHeaders()) | ||||||
|     result = json.loads(response.consumeBody()) |     result = json.loads(response.consumeBody()) | ||||||
|     self.assertEqual(result['items'][0]['head']['name'], 'level2-item1') |     self.assertEqual(result['items'][0]['head']['name'], 'level2-item1') | ||||||
| 
 | 
 | ||||||
|  |  | ||||||
|  | @ -110,7 +110,6 @@ class Authenticator(DummyFolder): | ||||||
| 
 | 
 | ||||||
|     def login(self): |     def login(self): | ||||||
|         req = self.request |         req = self.request | ||||||
|         logger.debug('login: %s %s %s', self, req.getTraversalStack(), req['PATH_INFO']) |  | ||||||
|         #print('***', dir(req)) |         #print('***', dir(req)) | ||||||
|         state = util.rndstr() |         state = util.rndstr() | ||||||
|         nonce = util.rndstr() |         nonce = util.rndstr() | ||||||
|  | @ -205,14 +204,14 @@ class Authenticator(DummyFolder): | ||||||
| 
 | 
 | ||||||
|     def loadOidcProviderData(self, force=False): |     def loadOidcProviderData(self, force=False): | ||||||
|         if config.oidc_provider.startswith('test'): |         if config.oidc_provider.startswith('test'): | ||||||
|             return |             pass | ||||||
|         if force or self.params.get('op_uris') is None: |         if force or self.params.get('op_uris') is None: | ||||||
|             uris = self.params['op_uris'] = {} |             uris = self.params['op_uris'] = {} | ||||||
|             opData = requests.get(self.params['op_config_url']).json() |             opData = requests.get(self.params['op_config_url']).json() | ||||||
|             for key in self.oidcProviderUris: |             for key in self.oidcProviderUris: | ||||||
|                 uris[key] = opData[key] |                 uris[key] = opData[key] | ||||||
|         if force or self.params.get('op_keys') is None: |         if force or self.params.get('op_keys') is None: | ||||||
|             self.params['op_keys'] = requests.get(uris['jwks_uri']).json() |             self.params['op_keys'] = requests.get(uris['jwks_uri']).json()['keys'] | ||||||
| 
 | 
 | ||||||
| 
 | 
 | ||||||
| @register('auth', Root) | @register('auth', Root) | ||||||
|  |  | ||||||
		Loading…
	
	Add table
		
		Reference in a new issue