Merge pull request #269 from Mognom/secure_processors

Secure processors improvements

src/wirecloud/proxy/processors.py

@@ -101,8 +101,8 @@ def process_secure_data(text, request, component_id, component_type):
 
  action = options.get('action', 'data')
  if action == 'data':
- substr = options.get('substr', '')
  var_ref = options.get('var_ref', '')
+ substr = options.get('substr', '{' + var_ref + '}')
  check_empty_params(substr=substr, var_ref=var_ref)
 
  value = get_variable_value_by_ref(var_ref, request['user'], cache_manager, component_id, component_type)
@@ -121,6 +121,26 @@ def process_secure_data(text, request, component_id, component_type):
  request['headers']['content-length'] = "%s" % len(new_body)
  request['data'] = BytesIO(new_body)
 
+ elif action == 'header':
+ var_ref = options.get('var_ref', '')
+ substr = options.get('substr', '{' + var_ref + '}')
+ header = options.get('header', '').lower()
+ check_empty_params(substr=substr, var_ref=var_ref, header=header)
+
+ value = get_variable_value_by_ref(var_ref, request['user'], cache_manager, component_id, component_type)
+ check_invalid_refs(var_ref=value)
+
+ encoding = options.get('encoding', 'none')
+ substr = substr.encode('utf8')
+ if encoding == 'url':
+ value = urlquote(value).encode('utf8')
+ elif encoding == 'base64':
+ value = base64.b64encode(value.encode('utf8'))[:-1]
+ else:
+ value = value.encode('utf8')
+
+ request['headers'][header] = request['headers'][header].replace(substr, value)
+
  elif action == 'basic_auth':
 
  user_ref = options.get('user_ref', '')

src/wirecloud/proxy/tests.py

@@ -552,6 +552,35 @@ def echo_response(method, url, *args, **kwargs):
  self.assertEqual(response.status_code, 200)
  self.assertEqual(self.read_response(response), b'username=|username|&password=dGVzdF9wYXNzd29yZA=')
 
+ def test_secure_data_default_substr(self):
+ user = User.objects.get(username='test')
+ iwidget = IWidget.objects.get(pk=1)
+ iwidget.set_variable_value('password', 'test_password', user)
+ iwidget.save()
+ self.assertNotEqual(iwidget.variables['password'], 'test_password')
+
+ self.client.login(username='test', password='test')
+
+ def echo_response(method, url, *args, **kwargs):
+ return {'status_code': 200, 'content': kwargs['data'].read()}
+
+ self.network._servers['http']['example.com'].add_response('POST', '/path', echo_response)
+ pass_ref = 'password'
+ user_ref = 'username'
+ secure_data_header = 'action=data, var_ref=' + pass_ref
+ secure_data_header += '&action=data, var_ref=' + user_ref
+ response = self.client.post(self.basic_url,
+ 'username={username}&password={password}',
+ content_type='application/x-www-form-urlencoded',
+ HTTP_HOST='localhost',
+ HTTP_REFERER='http://localhost/test/workspace',
+ HTTP_X_WIRECLOUD_SECURE_DATA=secure_data_header,
+ HTTP_WIRECLOUD_COMPONENT_TYPE="widget",
+ HTTP_WIRECLOUD_COMPONENT_ID="1")
+
+ self.assertEqual(response.status_code, 200)
+ self.assertEqual(self.read_response(response), b'username=test_username&password=test_password')
+
  def check_invalid_ref(self, invalid_ref):
 
  secure_data_header = 'action=data, substr=|password|, var_ref=' + invalid_ref
@@ -645,3 +674,130 @@ def echo_response(method, url, *args, **kwargs):
  HTTP_X_WIRECLOUD_SECURE_DATA=secure_data_header)
 
  self.assertEqual(response.status_code, 422)
+
+ def test_secure_data_header(self):
+ pass_ref = 'pref_secure'
+ self.client.login(username='test', password='test')
+
+ def echo_response(method, url, *args, **kwargs):
+ return {'status_code': 200, 'headers': kwargs['headers'], 'content': kwargs['data'].read()}
+
+ self.network._servers['http']['example.com'].add_response('POST', '/path', echo_response)
+
+ replaceHeader = "words {password}"
+ secure_data_header = 'action=header, header=headername, substr={password}, var_ref=' + pass_ref
+
+ response = self.client.post(self.basic_url,
+ 'username=|username|&password=|password|',
+ content_type='application/x-www-form-urlencoded',
+ HTTP_HEADERNAME=replaceHeader,
+ HTTP_HOST='localhost',
+ HTTP_REFERER='http://localhost/test/workspaceSecure',
+ HTTP_X_WIRECLOUD_SECURE_DATA=secure_data_header,
+ HTTP_WIRECLOUD_COMPONENT_TYPE="operator",
+ HTTP_WIRECLOUD_COMPONENT_ID="2")
+
+ self.assertEqual(response.status_code, 200)
+ self.assertEqual(self.get_response_headers(response)["headername"], "words test_password")
+ self.assertEqual(self.read_response(response), b'username=|username|&password=|password|')
+
+ def test_secure_data_header_concatenated(self):
+ pass_ref = 'pref_secure'
+ user_ref = 'username'
+ self.client.login(username='test', password='test')
+
+ def echo_response(method, url, *args, **kwargs):
+ return {'status_code': 200, 'headers': kwargs['headers'], 'content': kwargs['data'].read()}
+
+ self.network._servers['http']['example.com'].add_response('POST', '/path', echo_response)
+
+ replaceHeader = "words {username}:{password}"
+ secure_data_header = 'action=header, header=headername, substr={password}, var_ref=' + pass_ref + '&action=header, header=headername, substr={username}, var_ref=' + user_ref
+
+ response = self.client.post(self.basic_url,
+ 'username=|username|&password=|password|',
+ content_type='application/x-www-form-urlencoded',
+ HTTP_HEADERNAME=replaceHeader,
+ HTTP_HOST='localhost',
+ HTTP_REFERER='http://localhost/test/workspaceSecure',
+ HTTP_X_WIRECLOUD_SECURE_DATA=secure_data_header,
+ HTTP_WIRECLOUD_COMPONENT_TYPE="operator",
+ HTTP_WIRECLOUD_COMPONENT_ID="2")
+
+ self.assertEqual(response.status_code, 200)
+ self.assertEqual(self.get_response_headers(response)["headername"], "words test_username:test_password")
+ self.assertEqual(self.read_response(response), b'username=|username|&password=|password|')
+
+ def test_secure_data_header_default_substr(self):
+ pass_ref = 'pref_secure'
+ self.client.login(username='test', password='test')
+
+ def echo_response(method, url, *args, **kwargs):
+ return {'status_code': 200, 'headers': kwargs['headers'], 'content': kwargs['data'].read()}
+
+ self.network._servers['http']['example.com'].add_response('POST', '/path', echo_response)
+
+ replaceHeader = "words {pref_secure}"
+ secure_data_header = 'action=header, header=Headername, var_ref=' + pass_ref
+
+ response = self.client.post(self.basic_url,
+ 'username=|username|&password=|password|',
+ content_type='application/x-www-form-urlencoded',
+ HTTP_HEADERNAME=replaceHeader,
+ HTTP_HOST='localhost',
+ HTTP_REFERER='http://localhost/test/workspaceSecure',
+ HTTP_X_WIRECLOUD_SECURE_DATA=secure_data_header,
+ HTTP_WIRECLOUD_COMPONENT_TYPE="operator",
+ HTTP_WIRECLOUD_COMPONENT_ID="2")
+
+ self.assertEqual(response.status_code, 200)
+ self.assertEqual(self.get_response_headers(response)["headername"], "words test_password")
+ self.assertEqual(self.read_response(response), b'username=|username|&password=|password|')
+
+ def test_secure_data_header_missing_parameters(self):
+ pass_ref = 'pref_secure'
+ self.client.login(username='test', password='test')
+
+ def echo_response(method, url, *args, **kwargs):
+ return {'status_code': 200, 'headers': kwargs['headers'], 'content': kwargs['data'].read()}
+
+ self.network._servers['http']['example.com'].add_response('POST', '/path', echo_response)
+
+ replaceHeader = "words {pass_ref}"
+ secure_data_header = 'action=header, var_ref=' + pass_ref
+
+ response = self.client.post(self.basic_url,
+ 'username=|username|&password=|password|',
+ content_type='application/x-www-form-urlencoded',
+ HTTP_HEADERNAME=replaceHeader,
+ HTTP_HOST='localhost',
+ HTTP_REFERER='http://localhost/test/workspaceSecure',
+ HTTP_X_WIRECLOUD_SECURE_DATA=secure_data_header,
+ HTTP_WIRECLOUD_COMPONENT_TYPE="operator",
+ HTTP_WIRECLOUD_COMPONENT_ID="2")
+
+ self.assertEqual(response.status_code, 422)
+
+ def test_secure_data_header_empty_parameters(self):
+ pass_ref = 'pref_secure'
+ self.client.login(username='test', password='test')
+
+ def echo_response(method, url, *args, **kwargs):
+ return {'status_code': 200, 'headers': kwargs['headers'], 'content': kwargs['data'].read()}
+
+ self.network._servers['http']['example.com'].add_response('POST', '/path', echo_response)
+
+ replaceHeader = "words {pass_ref}"
+ secure_data_header = 'action=header, header='', var_ref=' + pass_ref
+
+ response = self.client.post(self.basic_url,
+ 'username=|username|&password=|password|',
+ content_type='application/x-www-form-urlencoded',
+ HTTP_HEADERNAME=replaceHeader,
+ HTTP_HOST='localhost',
+ HTTP_REFERER='http://localhost/test/workspaceSecure',
+ HTTP_X_WIRECLOUD_SECURE_DATA=secure_data_header,
+ HTTP_WIRECLOUD_COMPONENT_TYPE="operator",
+ HTTP_WIRECLOUD_COMPONENT_ID="2")
+
+ self.assertEqual(response.status_code, 422)