Skip to content

erp5
erp5
Commit 79b5a1ec authored by Jérome Perrin's avatar Jérome Perrin
Browse files
Options

Revert Fix AccessToken login with ERP5 Login 

This revert https://lab.nexedi.com/nexedi/erp5-capago/merge_requests/37
which was a backport of an intermediate state of
nexedi/erp5!838

Revert this first backport to easily backport again the latest merged
version of nexedi/erp5!838
parent 0229018e
Hide whitespace changes
Inline Side-by-side
Showing with 710 additions and 174 deletions
bt5/erp5_access_token/PortalTypePropertySheetTemplateItem/property_sheet_list.xml
View file @ 79b5a1ec
...@@ -6,7 +6,4 @@ ...@@ -6,7 +6,4 @@
<item>Reference</item> <item>Reference</item>
<item>Url</item> <item>Url</item>
</portal_type> </portal_type>
<portal_type id="Template Tool">
<item>TemplateToolERP5AccessTokenExtractionPluginConstraint</item>
</portal_type>
</property_sheet_list> </property_sheet_list>
\ No newline at end of file
bt5/erp5_access_token/SkinTemplateItem/portal_skins/erp5_access_token/AccessToken_getUserId.xml → bt5/erp5_access_token/SkinTemplateItem/portal_skins/erp5_access_token/AccessToken_getExternalLogin.xml
View file @ 79b5a1ec
...@@ -54,7 +54,7 @@ ...@@ -54,7 +54,7 @@
</item> </item>
<item> <item>
<key> <string>id</string> </key> <key> <string>id</string> </key>
<value> <string>AccessToken_getUserId</string> </value> <value> <string>AccessToken_getExternalLogin</string> </value>
</item> </item>
</dictionary> </dictionary>
</pickle> </pickle>
......
bt5/erp5_access_token/SkinTemplateItem/portal_skins/erp5_access_token/OneTimeRestrictedAccessToken_getUserId.xml → bt5/erp5_access_token/SkinTemplateItem/portal_skins/erp5_access_token/OneTimeRestrictedAccessToken_getExternalLogin.xml
View file @ 79b5a1ec
...@@ -54,7 +54,7 @@ ...@@ -54,7 +54,7 @@
</item> </item>
<item> <item>
<key> <string>id</string> </key> <key> <string>id</string> </key>
<value> <string>OneTimeRestrictedAccessToken_getUserId</string> </value> <value> <string>OneTimeRestrictedAccessToken_getExternalLogin</string> </value>
</item> </item>
</dictionary> </dictionary>
</pickle> </pickle>
......
bt5/erp5_access_token/SkinTemplateItem/portal_skins/erp5_access_token/RestrictedAccessToken_getUserId.py → bt5/erp5_access_token/SkinTemplateItem/portal_skins/erp5_access_token/RestrictedAccessToken_getExternalLogin.py
View file @ 79b5a1ec
...@@ -19,24 +19,9 @@ if access_token_document.getValidationState() == 'validated': ...@@ -19,24 +19,9 @@ if access_token_document.getValidationState() == 'validated':
# use hmac.compare_digest and not string comparison to avoid timing attacks # use hmac.compare_digest and not string comparison to avoid timing attacks
if not hmac.compare_digest(access_token_document.getReference(), reference): if not hmac.compare_digest(access_token_document.getReference(), reference):
return None return None
agent_document = access_token_document.getAgentValue() agent_document = access_token_document.getAgentValue()
if agent_document is not None: if agent_document is not None:
if agent_document.getPortalType() == 'Person':
# if this is a token for a person, only make accept if person has valid
# assignments (for compatibility with login/password authentication)
if agent_document.getValidationState() == 'deleted':
return None
now = DateTime()
for assignment in agent_document.contentValues(portal_type='Assignment'):
if assignment.getValidationState() == "open" and (
not assignment.hasStartDate() or assignment.getStartDate() <= now
) and (
not assignment.hasStopDate() or assignment.getStopDate() >= now
):
break
else:
return None
result = agent_document.Person_getUserId() result = agent_document.Person_getUserId()
return result return result
bt5/erp5_access_token/SkinTemplateItem/portal_skins/erp5_access_token/RestrictedAccessToken_getUserId.xml → bt5/erp5_access_token/SkinTemplateItem/portal_skins/erp5_access_token/RestrictedAccessToken_getExternalLogin.xml
View file @ 79b5a1ec
...@@ -54,7 +54,7 @@ ...@@ -54,7 +54,7 @@
</item> </item>
<item> <item>
<key> <string>id</string> </key> <key> <string>id</string> </key>
<value> <string>RestrictedAccessToken_getUserId</string> </value> <value> <string>RestrictedAccessToken_getExternalLogin</string> </value>
</item> </item>
</dictionary> </dictionary>
</pickle> </pickle>
......
bt5/erp5_access_token/SkinTemplateItem/portal_skins/erp5_access_token/RestrictedAccessToken_init.py
View file @ 79b5a1ec
alpha = '0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ' alpha = '0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ'
random_id = '' random_id = ''
for _ in range(0, 128): for i in range(0,128):
random_id += random.choice(alpha) random_id += random.choice(alpha)
# Define Reference from ID provided by portal_ids # Define Reference from ID provided by portal_ids
......
bt5/erp5_access_token/SkinTemplateItem/portal_skins/erp5_access_token/TemplateTool_checkERP5AccessTokenExtractionPluginExistenceConsistency.py deleted 100644 → 0
View file @ 0229018e
acl_users = context.getPortalObject().acl_users
token_extraction_id = "erp5_access_token_plugin"
access_token_plugin_list = [
plugin for plugin in acl_users.objectValues()
if plugin.meta_type == 'ERP5 Access Token Extraction Plugin']
if len(access_token_plugin_list) > 1:
return ["More than one plugin found: %s" % access_token_plugin_list]
error_list = []
if not access_token_plugin_list:
# A dumb http extraction plugin is required as fallback if we use an access token
# since https://github.com/Nexedi/erp5/commit/0bee523da0075c6efe3c06296dddd01d9dd5045a
# we enable it automatically at site creation, but for compatibility with old instances
# make sure it is created if needed
if 'erp5_dumb_http_extraction' not in acl_users.objectIds():
error_list.append("erp5_dumb_http_extraction is missing")
if fixit:
dispacher = acl_users.manage_addProduct['ERP5Security']
dispacher.addERP5DumbHTTPExtractionPlugin('erp5_dumb_http_extraction')
acl_users.erp5_dumb_http_extraction.manage_activateInterfaces(('IExtractionPlugin', ))
error_list.append("erp5_access_token_plugin is missing")
if fixit:
dispacher = acl_users.manage_addProduct['ERP5Security']
dispacher.addERP5AccessTokenExtractionPlugin(token_extraction_id)
access_token_plugin_list = [getattr(acl_users, token_extraction_id)]
if access_token_plugin_list:
access_token_plugin, = access_token_plugin_list
# We only check that our plugin is enabled for IAuthenticationPlugin, this covers both
# cases where plugin was not enabled at all or was enabled only for IExtractionPlugin
IAuthenticationPlugin = [
# Products.PluggableAuthService.interfaces.plugins.IAuthenticationPlugin cannot
# be imported in restricted python but we can get it this way.
x for x in acl_users.plugins.listPluginTypeInfo()
if x['id'] == 'IAuthenticationPlugin'][0]['interface']
if (access_token_plugin.getId()
not in acl_users.plugins.listPluginIds(IAuthenticationPlugin)):
error_list.append("erp5_access_token_plugin is not activated")
if fixit:
access_token_plugin.manage_activateInterfaces((
'IExtractionPlugin',
'IAuthenticationPlugin',))
return error_list
bt5/erp5_access_token/SkinTemplateItem/portal_skins/erp5_access_token/TemplateTool_checkERP5AccessTokenExtractionPluginExistenceConsistency.xml deleted 100644 → 0
View file @ 0229018e
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="PythonScript" module="Products.PythonScripts.PythonScript"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>Script_magic</string> </key>
<value> <int>3</int> </value>
</item>
<item>
<key> <string>_bind_names</string> </key>
<value>
<object>
<klass>
<global name="NameAssignments" module="Shared.DC.Scripts.Bindings"/>
</klass>
<tuple/>
<state>
<dictionary>
<item>
<key> <string>_asgns</string> </key>
<value>
<dictionary>
<item>
<key> <string>name_container</string> </key>
<value> <string>container</string> </value>
</item>
<item>
<key> <string>name_context</string> </key>
<value> <string>context</string> </value>
</item>
<item>
<key> <string>name_m_self</string> </key>
<value> <string>script</string> </value>
</item>
<item>
<key> <string>name_subpath</string> </key>
<value> <string>traverse_subpath</string> </value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</state>
</object>
</value>
</item>
<item>
<key> <string>_params</string> </key>
<value> <string>fixit=False</string> </value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>TemplateTool_checkERP5AccessTokenExtractionPluginExistenceConsistency</string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
bt5/erp5_access_token/TestTemplateItem/portal_components/test.erp5.testERP5AccessTokenAlarm.py 0 → 100644
View file @ 79b5a1ec
# Copyright (c) 2002-2013 Nexedi SA and Contributors. All Rights Reserved.
from DateTime import DateTime
from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
class TestERP5AccessTokenAlarm(ERP5TypeTestCase):
def getBusinessTemplateList(self):
return ('erp5_base',
'erp5_access_token')
def test_alarm_old_validated_restricted_access_token(self):
access_token = self.portal.access_token_module.newContent(
portal_type="One Time Restricted Access Token",
)
access_token.workflow_history['edit_workflow'] = [{
'comment':'Fake history',
'error_message': '',
'actor': 'ERP5TypeTestCase',
'state': 'current',
'time': DateTime('2012/11/15 11:11'),
'action': 'foo_action'
}]
self.portal.portal_workflow._jumpToStateFor(access_token, 'validated')
self.tic()
self.portal.portal_alarms.\
erp5_garbage_collect_one_time_restricted_access_token.activeSense()
self.tic()
self.assertEqual('invalidated', access_token.getValidationState())
self.assertEqual(
'Unused for 1 day.',
access_token.workflow_history['validation_workflow'][-1]['comment'])
def test_alarm_recent_validated_restricted_access_token(self):
access_token = self.portal.access_token_module.newContent(
portal_type="One Time Restricted Access Token",
)
self.portal.portal_workflow._jumpToStateFor(access_token, 'validated')
self.tic()
self.portal.portal_alarms.\
erp5_garbage_collect_one_time_restricted_access_token.activeSense()
self.tic()
self.assertEqual('validated', access_token.getValidationState())
def test_alarm_old_non_validated_restricted_access_token(self):
access_token = self.portal.access_token_module.newContent(
portal_type="One Time Restricted Access Token",
)
access_token.workflow_history['edit_workflow'] = [{
'comment':'Fake history',
'error_message': '',
'actor': 'ERP5TypeTestCase',
'state': 'current',
'time': DateTime('2012/11/15 11:11'),
'action': 'foo_action'
}]
self.tic()
self.portal.portal_alarms.\
erp5_garbage_collect_one_time_restricted_access_token.activeSense()
self.tic()
self.assertEqual('draft', access_token.getValidationState())
bt5/erp5_access_token/TestTemplateItem/portal_components/test.erp5.testERP5AccessToken.xml → bt5/erp5_access_token/TestTemplateItem/portal_components/test.erp5.testERP5AccessTokenAlarm.xml
View file @ 79b5a1ec
...@@ -14,7 +14,7 @@ ...@@ -14,7 +14,7 @@
</item> </item>
<item> <item>
<key> <string>default_reference</string> </key> <key> <string>default_reference</string> </key>
<value> <string>testERP5AccessToken</string> </value> <value> <string>testERP5AccessTokenAlarm</string> </value>
</item> </item>
<item> <item>
<key> <string>description</string> </key> <key> <string>description</string> </key>
...@@ -24,7 +24,7 @@ ...@@ -24,7 +24,7 @@
</item> </item>
<item> <item>
<key> <string>id</string> </key> <key> <string>id</string> </key>
<value> <string>test.erp5.testERP5AccessToken</string> </value> <value> <string>test.erp5.testERP5AccessTokenAlarm</string> </value>
</item> </item>
<item> <item>
<key> <string>portal_type</string> </key> <key> <string>portal_type</string> </key>
......
bt5/erp5_access_token/TestTemplateItem/portal_components/test.erp5.testERP5AccessToken.py → bt5/erp5_access_token/TestTemplateItem/portal_components/test.erp5.testERP5AccessTokenSkins.py
View file @ 79b5a1ec
# -*- coding: utf-8 -*- # Copyright (c) 2002-2013 Nexedi SA and Contributors. All Rights Reserved.
##############################################################################
#
# Copyright (c) 2002-2019 Nexedi SA and Contributors. All Rights Reserved.
# Tristan Cavelier <tristan.cavelier@nexedi.com>
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsability of assessing all potential
# consequences resulting from its eventual inadequacies and bugs
# End users who are looking for a ready-to-use solution with commercial
# garantees and support are strongly adviced to contract a Free Software
# Service Company
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#
##############################################################################
from ZPublisher.HTTPRequest import HTTPRequest
from ZPublisher.HTTPResponse import HTTPResponse
from Products.PluggableAuthService.interfaces.plugins import IAuthenticationPlugin
from DateTime import DateTime
import base64
import StringIO
from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
from Products.ERP5Security.ERP5DumbHTTPExtractionPlugin import ERP5DumbHTTPExtractionPlugin import transaction
class TestERP5AccessTokenSkins(ERP5TypeTestCase):
test_token_extraction_id = 'test_erp5_access_token_extraction'
class AccessTokenTestCase(ERP5TypeTestCase):
def getBusinessTemplateList(self): def getBusinessTemplateList(self):
return ('erp5_base', return ('erp5_base',
'erp5_access_token') 'erp5_access_token')
def _createPerson(self, new_id, password=None):
"""Creates a person in person module, and returns the object, after
indexing is done. """
person_module = self.getPersonModule()
person = person_module.newContent(portal_type='Person',
reference='TESTP-' + new_id)
if password:
person.setPassword(password)
person.newContent(portal_type = 'Assignment').open()
self.tic()
return person
class TestERP5AccessTokenSkins(AccessTokenTestCase):
def generateNewId(self): def generateNewId(self):
return str(self.portal.portal_ids.generateNewId( return str(self.portal.portal_ids.generateNewId(
id_group=('erp5_access_token_test_id'))) id_group=('erp5_access_token_test_id')))
...@@ -65,15 +19,40 @@ class TestERP5AccessTokenSkins(AccessTokenTestCase): ...@@ -65,15 +19,40 @@ class TestERP5AccessTokenSkins(AccessTokenTestCase):
""" """
This is ran before anything, used to set the environment This is ran before anything, used to set the environment
""" """
self.portal = self.getPortalObject()
self.new_id = self.generateNewId() self.new_id = self.generateNewId()
self.portal.portal_templates.TemplateTool_checkERP5AccessTokenExtractionPluginExistenceConsistency( self._setupAccessTokenExtraction()
fixit=True) transaction.commit()
self.tic() self.tic()
def _setupAccessTokenExtraction(self):
pas = self.portal.acl_users
access_extraction_list = [q for q in pas.objectValues() \
if q.meta_type == 'ERP5 Access Token Extraction Plugin']
if len(access_extraction_list) == 0:
dispacher = pas.manage_addProduct['ERP5Security']
dispacher.addERP5AccessTokenExtractionPlugin(self.test_token_extraction_id)
getattr(pas, self.test_token_extraction_id).manage_activateInterfaces(
('IExtractionPlugin',))
elif len(access_extraction_list) == 1:
self.test_token_extraction_id = access_extraction_list[0].getId()
elif len(access_extraction_list) > 1:
raise ValueError
transaction.commit()
def _createPerson(self, new_id):
"""Creates a person in person module, and returns the object, after
indexing is done. """
person_module = self.getPersonModule()
person = person_module.newContent(portal_type='Person',
user_id='TESTP-' + new_id)
person.newContent(portal_type = 'Assignment').open()
transaction.commit()
return person
def _getTokenCredential(self, request): def _getTokenCredential(self, request):
"""Authenticate the request and return (user_id, login) or None if not authorized.""" plugin = getattr(self.portal.acl_users, self.test_token_extraction_id)
plugin = self.portal.acl_users.erp5_access_token_plugin return plugin.extractCredentials(request)
return plugin.authenticateCredentials(plugin.extractCredentials(request))
def _createRestrictedAccessToken(self, new_id, person, method, url_string): def _createRestrictedAccessToken(self, new_id, person, method, url_string):
access_token = self.portal.access_token_module.newContent( access_token = self.portal.access_token_module.newContent(
...@@ -96,7 +75,7 @@ class TestERP5AccessTokenSkins(AccessTokenTestCase): ...@@ -96,7 +75,7 @@ class TestERP5AccessTokenSkins(AccessTokenTestCase):
return access_token return access_token
def test_working_token(self): def test_working_token(self):
person = self._createPerson(self.new_id) person = self.person = self._createPerson(self.new_id)
access_url = "http://exemple.com/foo" access_url = "http://exemple.com/foo"
access_method = "GET" access_method = "GET"
access_token = self._createRestrictedAccessToken(self.new_id, access_token = self._createRestrictedAccessToken(self.new_id,
...@@ -112,14 +91,10 @@ class TestERP5AccessTokenSkins(AccessTokenTestCase): ...@@ -112,14 +91,10 @@ class TestERP5AccessTokenSkins(AccessTokenTestCase):
self.portal.REQUEST.form["access_token_secret"] = access_token.getReference() self.portal.REQUEST.form["access_token_secret"] = access_token.getReference()
result = self._getTokenCredential(self.portal.REQUEST) result = self._getTokenCredential(self.portal.REQUEST)
self.assertTrue(result) self.assertEqual(result.get('external_login'), person.Person_getUserId())
user_id, login = result
self.assertEqual(user_id, person.Person_getUserId())
# tokens have a login value, for auditing purposes
self.assertIn('token', login)
def test_bad_token(self): def test_bad_token(self):
person = self._createPerson(self.new_id) person = self.person = self._createPerson(self.new_id)
access_url = "http://exemple.com/foo" access_url = "http://exemple.com/foo"
access_method = "GET" access_method = "GET"
access_token = self._createRestrictedAccessToken(self.new_id, access_token = self._createRestrictedAccessToken(self.new_id,
...@@ -135,32 +110,10 @@ class TestERP5AccessTokenSkins(AccessTokenTestCase): ...@@ -135,32 +110,10 @@ class TestERP5AccessTokenSkins(AccessTokenTestCase):
self.portal.REQUEST.form["access_token_secret"] = access_token.getReference() self.portal.REQUEST.form["access_token_secret"] = access_token.getReference()
result = self._getTokenCredential(self.portal.REQUEST) result = self._getTokenCredential(self.portal.REQUEST)
self.assertFalse(result) self.assertEqual(result, {})
def test_token_without_assignment(self):
# Token does not work when person has no open assignment
person = self._createPerson(self.new_id)
for assignment in person.contentValues(portal_type='Assignment'):
assignment.close()
access_url = "http://exemple.com/foo"
access_method = "GET"
access_token = self._createRestrictedAccessToken(self.new_id,
person,
access_method,
access_url)
access_token.validate()
self.tic()
self.portal.REQUEST.form["access_token"] = access_token.getId()
self.portal.REQUEST["REQUEST_METHOD"] = access_method
self.portal.REQUEST["ACTUAL_URL"] = access_url
self.portal.REQUEST.form["access_token_secret"] = access_token.getReference()
result = self._getTokenCredential(self.portal.REQUEST)
self.assertFalse(result)
def test_RestrictedAccessToken_getUserId(self): def test_RestrictedAccessToken_getExternalLogin(self):
person = self._createPerson(self.new_id) person = self.person = self._createPerson(self.new_id)
access_url = "http://exemple.com/foo" access_url = "http://exemple.com/foo"
access_method = "GET" access_method = "GET"
access_token = self._createRestrictedAccessToken(self.new_id, access_token = self._createRestrictedAccessToken(self.new_id,
...@@ -174,13 +127,13 @@ class TestERP5AccessTokenSkins(AccessTokenTestCase): ...@@ -174,13 +127,13 @@ class TestERP5AccessTokenSkins(AccessTokenTestCase):
self.portal.REQUEST["ACTUAL_URL"] = access_url self.portal.REQUEST["ACTUAL_URL"] = access_url
self.portal.REQUEST.form["access_token_secret"] = access_token.getReference() self.portal.REQUEST.form["access_token_secret"] = access_token.getReference()
result = access_token.RestrictedAccessToken_getUserId() result = access_token.RestrictedAccessToken_getExternalLogin()
self.assertEqual(result, person.Person_getUserId()) self.assertEqual(result, person.Person_getUserId())
self.assertEqual(access_token.getValidationState(), 'validated') self.assertEqual(access_token.getValidationState(), 'validated')
def test_RestrictedAccessToken_getUserId_access_token_secret(self): def test_RestrictedAccessToken_getExternalLogin_access_token_secret(self):
person = self._createPerson(self.new_id) person = self.person = self._createPerson(self.new_id)
access_url = "http://exemple.com/foo" access_url = "http://exemple.com/foo"
access_method = "GET" access_method = "GET"
access_token = self._createRestrictedAccessToken(self.new_id, access_token = self._createRestrictedAccessToken(self.new_id,
...@@ -193,7 +146,7 @@ class TestERP5AccessTokenSkins(AccessTokenTestCase): ...@@ -193,7 +146,7 @@ class TestERP5AccessTokenSkins(AccessTokenTestCase):
self.portal.REQUEST["REQUEST_METHOD"] = access_method self.portal.REQUEST["REQUEST_METHOD"] = access_method
self.portal.REQUEST["ACTUAL_URL"] = access_url self.portal.REQUEST["ACTUAL_URL"] = access_url
result = access_token.RestrictedAccessToken_getUserId() result = access_token.RestrictedAccessToken_getExternalLogin()
self.assertEqual(result, None) self.assertEqual(result, None)
self.portal.REQUEST.form["access_token_secret"] = "XYXYXYXY" self.portal.REQUEST.form["access_token_secret"] = "XYXYXYXY"
...@@ -201,12 +154,12 @@ class TestERP5AccessTokenSkins(AccessTokenTestCase): ...@@ -201,12 +154,12 @@ class TestERP5AccessTokenSkins(AccessTokenTestCase):
self.portal.REQUEST.form["access_token_secret"] = access_token.getReference() self.portal.REQUEST.form["access_token_secret"] = access_token.getReference()
result = access_token.RestrictedAccessToken_getUserId() result = access_token.RestrictedAccessToken_getExternalLogin()
self.assertEqual(result, person.Person_getUserId()) self.assertEqual(result, person.Person_getUserId())
self.assertEqual(access_token.getValidationState(), 'validated') self.assertEqual(access_token.getValidationState(), 'validated')
def test_RestrictedAccessToken_getUserId_no_agent(self): def test_RestrictedAccessToken_getExternalLogin_no_agent(self):
access_url = "http://exemple.com/foo" access_url = "http://exemple.com/foo"
access_method = "GET" access_method = "GET"
access_token = self._createRestrictedAccessToken(self.new_id, access_token = self._createRestrictedAccessToken(self.new_id,
...@@ -220,11 +173,11 @@ class TestERP5AccessTokenSkins(AccessTokenTestCase): ...@@ -220,11 +173,11 @@ class TestERP5AccessTokenSkins(AccessTokenTestCase):
self.portal.REQUEST["ACTUAL_URL"] = access_url self.portal.REQUEST["ACTUAL_URL"] = access_url
self.portal.REQUEST.form["access_token_secret"] = access_token.getReference() self.portal.REQUEST.form["access_token_secret"] = access_token.getReference()
result = access_token.RestrictedAccessToken_getUserId() result = access_token.RestrictedAccessToken_getExternalLogin()
self.assertEqual(result, None) self.assertEqual(result, None)
def test_RestrictedAccessToken_getUserId_wrong_values(self): def test_RestrictedAccessToken_getExternalLogin_wrong_values(self):
person = self._createPerson(self.new_id) person = self.person = self._createPerson(self.new_id)
access_url = "http://exemple.com/foo" access_url = "http://exemple.com/foo"
access_method = "GET" access_method = "GET"
access_token = self._createRestrictedAccessToken(self.new_id, access_token = self._createRestrictedAccessToken(self.new_id,
...@@ -232,7 +185,7 @@ class TestERP5AccessTokenSkins(AccessTokenTestCase): ...@@ -232,7 +185,7 @@ class TestERP5AccessTokenSkins(AccessTokenTestCase):
access_method, access_method,
access_url) access_url)
self.tic() self.tic()
result = access_token.RestrictedAccessToken_getUserId() result = access_token.RestrictedAccessToken_getExternalLogin()
self.assertEqual(result, None) self.assertEqual(result, None)
access_token.validate() access_token.validate()
...@@ -242,23 +195,23 @@ class TestERP5AccessTokenSkins(AccessTokenTestCase): ...@@ -242,23 +195,23 @@ class TestERP5AccessTokenSkins(AccessTokenTestCase):
self.portal.REQUEST["ACTUAL_URL"] = access_url self.portal.REQUEST["ACTUAL_URL"] = access_url
self.portal.REQUEST.form["access_token_secret"] = access_token.getReference() self.portal.REQUEST.form["access_token_secret"] = access_token.getReference()
result = access_token.RestrictedAccessToken_getUserId() result = access_token.RestrictedAccessToken_getExternalLogin()
self.assertEqual(result, None) self.assertEqual(result, None)
self.portal.REQUEST["ACTUAL_URL"] = "http://exemple.com/foo.bar" self.portal.REQUEST["ACTUAL_URL"] = "http://exemple.com/foo.bar"
result = access_token.RestrictedAccessToken_getUserId() result = access_token.RestrictedAccessToken_getExternalLogin()
self.assertEqual(result, None) self.assertEqual(result, None)
access_token.invalidate() access_token.invalidate()
self.tic() self.tic()
result = access_token.RestrictedAccessToken_getUserId() result = access_token.RestrictedAccessToken_getExternalLogin()
self.assertEqual(result, None) self.assertEqual(result, None)
def test_OneTimeRestrictedAccessToken_getUserId(self): def test_OneTimeRestrictedAccessToken_getExternalLogin(self):
person = self._createPerson(self.new_id) person = self.person = self._createPerson(self.new_id)
access_url = "http://exemple.com/foo" access_url = "http://exemple.com/foo"
access_method = "GET" access_method = "GET"
access_token = self._createOneTimeRestrictedAccessToken(self.new_id, access_token = self._createOneTimeRestrictedAccessToken(self.new_id,
...@@ -271,13 +224,13 @@ class TestERP5AccessTokenSkins(AccessTokenTestCase): ...@@ -271,13 +224,13 @@ class TestERP5AccessTokenSkins(AccessTokenTestCase):
self.portal.REQUEST["REQUEST_METHOD"] = access_method self.portal.REQUEST["REQUEST_METHOD"] = access_method
self.portal.REQUEST["ACTUAL_URL"] = access_url self.portal.REQUEST["ACTUAL_URL"] = access_url
result = access_token.OneTimeRestrictedAccessToken_getUserId() result = access_token.OneTimeRestrictedAccessToken_getExternalLogin()
self.assertEqual(result, person.Person_getUserId()) self.assertEqual(result, person.Person_getUserId())
self.assertEqual(access_token.getValidationState(), 'invalidated') self.assertEqual(access_token.getValidationState(), 'invalidated')
def test_OneTimeRestrictedAccessToken_getUserId_wrong_values(self): def test_OneTimeRestrictedAccessToken_getExternalLogin_wrong_values(self):
person = self._createPerson(self.new_id) person = self.person = self._createPerson(self.new_id)
access_url = "http://exemple.com/foo" access_url = "http://exemple.com/foo"
access_method = "POST" access_method = "POST"
access_token = self._createOneTimeRestrictedAccessToken(self.new_id, access_token = self._createOneTimeRestrictedAccessToken(self.new_id,
...@@ -285,7 +238,7 @@ class TestERP5AccessTokenSkins(AccessTokenTestCase): ...@@ -285,7 +238,7 @@ class TestERP5AccessTokenSkins(AccessTokenTestCase):
access_method, access_method,
access_url) access_url)
self.tic() self.tic()
result = access_token.OneTimeRestrictedAccessToken_getUserId() result = access_token.OneTimeRestrictedAccessToken_getExternalLogin()
self.assertEqual(result, None) self.assertEqual(result, None)
access_token.validate() access_token.validate()
...@@ -294,153 +247,10 @@ class TestERP5AccessTokenSkins(AccessTokenTestCase): ...@@ -294,153 +247,10 @@ class TestERP5AccessTokenSkins(AccessTokenTestCase):
self.portal.REQUEST["REQUEST_METHOD"] = "GET" self.portal.REQUEST["REQUEST_METHOD"] = "GET"
self.portal.REQUEST["ACTUAL_URL"] = access_url self.portal.REQUEST["ACTUAL_URL"] = access_url
result = access_token.OneTimeRestrictedAccessToken_getUserId() result = access_token.OneTimeRestrictedAccessToken_getExternalLogin()
self.assertEqual(result, None) self.assertEqual(result, None)
self.portal.REQUEST["ACTUAL_URL"] = "http://exemple.com/foo.bar" self.portal.REQUEST["ACTUAL_URL"] = "http://exemple.com/foo.bar"
result = access_token.OneTimeRestrictedAccessToken_getUserId() result = access_token.OneTimeRestrictedAccessToken_getExternalLogin()
self.assertEqual(result, None) self.assertEqual(result, None)
class TestERP5AccessTokenAlarm(AccessTokenTestCase):
def test_alarm_old_validated_restricted_access_token(self):
access_token = self.portal.access_token_module.newContent(
portal_type="One Time Restricted Access Token",
)
access_token.workflow_history['edit_workflow'] = [{
'comment':'Fake history',
'error_message': '',
'actor': 'ERP5TypeTestCase',
'state': 'current',
'time': DateTime('2012/11/15 11:11'),
'action': 'foo_action'
}]
self.portal.portal_workflow._jumpToStateFor(access_token, 'validated')
self.tic()
self.portal.portal_alarms.\
erp5_garbage_collect_one_time_restricted_access_token.activeSense()
self.tic()
self.assertEqual('invalidated', access_token.getValidationState())
self.assertEqual(
'Unused for 1 day.',
access_token.workflow_history['validation_workflow'][-1]['comment'])
def test_alarm_recent_validated_restricted_access_token(self):
access_token = self.portal.access_token_module.newContent(
portal_type="One Time Restricted Access Token",
)
self.portal.portal_workflow._jumpToStateFor(access_token, 'validated')
self.tic()
self.portal.portal_alarms.\
erp5_garbage_collect_one_time_restricted_access_token.activeSense()
self.tic()
self.assertEqual('validated', access_token.getValidationState())
def test_alarm_old_non_validated_restricted_access_token(self):
access_token = self.portal.access_token_module.newContent(
portal_type="One Time Restricted Access Token",
)
access_token.workflow_history['edit_workflow'] = [{
'comment':'Fake history',
'error_message': '',
'actor': 'ERP5TypeTestCase',
'state': 'current',
'time': DateTime('2012/11/15 11:11'),
'action': 'foo_action'
}]
self.tic()
self.portal.portal_alarms.\
erp5_garbage_collect_one_time_restricted_access_token.activeSense()
self.tic()
self.assertEqual('draft', access_token.getValidationState())
class TestERP5DumbHTTPExtractionPlugin(AccessTokenTestCase):
test_id = 'test_erp5_dumb_http_extraction'
def generateNewId(self):
return str(self.portal.portal_ids.generateNewId(
id_group=('erp5_dumb_http_test_id')))
def afterSetUp(self):
"""
This is ran before anything, used to set the environment
"""
self.new_id = self.generateNewId()
self._setupDumbHTTPExtraction()
self.tic()
def do_fake_request(self, request_method, headers=None):
if headers is None:
headers = {}
__version__ = "0.1"
env={}
env['SERVER_NAME']='bobo.server'
env['SERVER_PORT']='80'
env['REQUEST_METHOD']=request_method
env['REMOTE_ADDR']='204.183.226.81 '
env['REMOTE_HOST']='bobo.remote.host'
env['HTTP_USER_AGENT']='Bobo/%s' % __version__
env['HTTP_HOST']='127.0.0.1'
env['SERVER_SOFTWARE']='Bobo/%s' % __version__
env['SERVER_PROTOCOL']='HTTP/1.0 '
env['HTTP_ACCEPT']='image/gif, image/x-xbitmap, image/jpeg, */* '
env['SERVER_HOSTNAME']='bobo.server.host'
env['GATEWAY_INTERFACE']='CGI/1.1 '
env['SCRIPT_NAME']='Main'
env.update(headers)
return HTTPRequest(StringIO.StringIO(), env, HTTPResponse())
def _setupDumbHTTPExtraction(self):
pas = self.portal.acl_users
access_extraction_list = [q for q in pas.objectValues() \
if q.meta_type == 'ERP5 Dumb HTTP Extraction Plugin']
if len(access_extraction_list) == 0:
dispacher = pas.manage_addProduct['ERP5Security']
dispacher.addERP5DumbHTTPExtractionPlugin(self.test_id)
getattr(pas, self.test_id).manage_activateInterfaces(
('IExtractionPlugin',))
elif len(access_extraction_list) == 1:
self.test_id = access_extraction_list[0].getId()
elif len(access_extraction_list) > 1:
raise ValueError
self.commit()
def test_working_authentication(self):
self._createPerson(self.new_id, "test")
request = self.do_fake_request("GET", {"HTTP_AUTHORIZATION": "Basic " + base64.b64encode("%s:test" % self.new_id)})
ret = ERP5DumbHTTPExtractionPlugin("default_extraction").extractCredentials(request)
self.assertEqual(ret, {'login': self.new_id, 'password': 'test', 'remote_host': 'bobo.remote.host', 'remote_address': '204.183.226.81 '})
class TestERP5AccessTokenUpgraderEnablePlugin(AccessTokenTestCase):
def afterSetUp(self):
# disable plugin if it had been enabled by another test.
acl_users = self.portal.acl_users
acl_users.manage_delObjects(ids=[
x.getId() for x in
acl_users.objectValues(spec=('ERP5 Access Token Extraction Plugin',))])
self.commit()
def test_post_upgrade_constraint_enable_plugin(self):
consistency_list = self.portal.portal_templates.checkConsistency(
filter={"constraint_type": "post_upgrade"})
self.assertIn(
'erp5_access_token_plugin is missing',
[x.message for x in consistency_list])
self.portal.portal_templates.checkConsistency(
fixit=True,
filter={"constraint_type": "post_upgrade"})
self.commit()
self.assertIn(
'erp5_access_token_plugin',
self.portal.acl_users.plugins.listPluginIds(IAuthenticationPlugin))
\ No newline at end of file
bt5/erp5_access_token/PropertySheetTemplateItem/portal_property_sheets/TemplateToolERP5AccessTokenExtractionPluginConstraint/ERP5AccessTokenExtractionPlugin_existence_constraint.xml → bt5/erp5_access_token/TestTemplateItem/portal_components/test.erp5.testERP5AccessTokenSkins.xml
View file @ 79b5a1ec
...@@ -2,47 +2,61 @@ ...@@ -2,47 +2,61 @@
<ZopeData> <ZopeData>
<record id="1" aka="AAAAAAAAAAE="> <record id="1" aka="AAAAAAAAAAE=">
<pickle> <pickle>
<global name="Script Constraint" module="erp5.portal_type"/> <global name="Test Component" module="erp5.portal_type"/>
</pickle> </pickle>
<pickle> <pickle>
<dictionary> <dictionary>
<item> <item>
<key> <string>_identity_criterion</string> </key> <key> <string>_recorded_property_dict</string> </key>
<value> <value>
<persistent> <string encoding="base64">AAAAAAAAAAI=</string> </persistent> <persistent> <string encoding="base64">AAAAAAAAAAI=</string> </persistent>
</value> </value>
</item> </item>
<item> <item>
<key> <string>_range_criterion</string> </key> <key> <string>default_reference</string> </key>
<value> <string>testERP5AccessTokenSkins</string> </value>
</item>
<item>
<key> <string>description</string> </key>
<value> <value>
<persistent> <string encoding="base64">AAAAAAAAAAM=</string> </persistent> <none/>
</value> </value>
</item> </item>
<item> <item>
<key> <string>categories</string> </key> <key> <string>id</string> </key>
<value> <string>test.erp5.testERP5AccessTokenSkins</string> </value>
</item>
<item>
<key> <string>portal_type</string> </key>
<value> <string>Test Component</string> </value>
</item>
<item>
<key> <string>sid</string> </key>
<value> <value>
<tuple> <none/>
<string>constraint_type/post_upgrade</string>
</tuple>
</value> </value>
</item> </item>
<item> <item>
<key> <string>description</string> </key> <key> <string>text_content_error_message</string> </key>
<value> <value>
<none/> <tuple/>
</value> </value>
</item> </item>
<item> <item>
<key> <string>id</string> </key> <key> <string>text_content_warning_message</string> </key>
<value> <string>ERP5AccessTokenExtractionPlugin_existence_constraint</string> </value> <value>
<tuple/>
</value>
</item> </item>
<item> <item>
<key> <string>portal_type</string> </key> <key> <string>version</string> </key>
<value> <string>Script Constraint</string> </value> <value> <string>erp5</string> </value>
</item> </item>
<item> <item>
<key> <string>script_id</string> </key> <key> <string>workflow_history</string> </key>
<value> <string>TemplateTool_checkERP5AccessTokenExtractionPluginExistenceConsistency</string> </value> <value>
<persistent> <string encoding="base64">AAAAAAAAAAM=</string> </persistent>
</value>
</item> </item>
</dictionary> </dictionary>
</pickle> </pickle>
...@@ -71,10 +85,39 @@ ...@@ -71,10 +85,39 @@
<item> <item>
<key> <string>data</string> </key> <key> <string>data</string> </key>
<value> <value>
<dictionary/> <dictionary>
<item>
<key> <string>component_validation_workflow</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAQ=</string> </persistent>
</value>
</item>
</dictionary>
</value> </value>
</item> </item>
</dictionary> </dictionary>
</pickle> </pickle>
</record> </record>
<record id="4" aka="AAAAAAAAAAQ=">
<pickle>
<global name="WorkflowHistoryList" module="Products.ERP5Type.patches.WorkflowTool"/>
</pickle>
<pickle>
<tuple>
<none/>
<list>
<dictionary>
<item>
<key> <string>action</string> </key>
<value> <string>validate</string> </value>
</item>
<item>
<key> <string>validation_state</string> </key>
<value> <string>validated</string> </value>
</item>
</dictionary>
</list>
</tuple>
</pickle>
</record>
</ZopeData> </ZopeData>
bt5/erp5_access_token/TestTemplateItem/portal_components/test.erp5.testERP5DumbHTTPExtractionPlugin.py 0 → 100644
View file @ 79b5a1ec
# -*- coding: utf-8 -*-
##############################################################################
#
# Copyright (c) 2015 Nexedi SA and Contributors. All Rights Reserved.
# Tristan Cavelier <tristan.cavelier@nexedi.com>
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsability of assessing all potential
# consequences resulting from its eventual inadequacies and bugs
# End users who are looking for a ready-to-use solution with commercial
# garantees and support are strongly adviced to contract a Free Software
# Service Company
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#
##############################################################################
from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
from ZPublisher.HTTPRequest import HTTPRequest
from ZPublisher.HTTPResponse import HTTPResponse
from Products.ERP5Security.ERP5DumbHTTPExtractionPlugin import ERP5DumbHTTPExtractionPlugin
import base64
import transaction
import StringIO
class TestERP5DumbHTTPExtractionPlugin(ERP5TypeTestCase):
test_id = 'test_erp5_dumb_http_extraction'
def getBusinessTemplateList(self):
return ('erp5_base',)
def generateNewId(self):
return str(self.portal.portal_ids.generateNewId(
id_group=('erp5_dumb_http_test_id')))
def afterSetUp(self):
"""
This is ran before anything, used to set the environment
"""
self.portal = self.getPortalObject()
self.new_id = self.generateNewId()
self._setupDumbHTTPExtraction()
transaction.commit()
self.tic()
def do_fake_request(self, request_method, headers={}):
__version__ = "0.1"
env={}
env['SERVER_NAME']='bobo.server'
env['SERVER_PORT']='80'
env['REQUEST_METHOD']=request_method
env['REMOTE_ADDR']='204.183.226.81 '
env['REMOTE_HOST']='bobo.remote.host'
env['HTTP_USER_AGENT']='Bobo/%s' % __version__
env['HTTP_HOST']='127.0.0.1'
env['SERVER_SOFTWARE']='Bobo/%s' % __version__
env['SERVER_PROTOCOL']='HTTP/1.0 '
env['HTTP_ACCEPT']='image/gif, image/x-xbitmap, image/jpeg, */* '
env['SERVER_HOSTNAME']='bobo.server.host'
env['GATEWAY_INTERFACE']='CGI/1.1 '
env['SCRIPT_NAME']='Main'
env.update(headers)
return HTTPRequest(StringIO.StringIO(), env, HTTPResponse())
def _setupDumbHTTPExtraction(self):
pas = self.portal.acl_users
access_extraction_list = [q for q in pas.objectValues() \
if q.meta_type == 'ERP5 Dumb HTTP Extraction Plugin']
if len(access_extraction_list) == 0:
dispacher = pas.manage_addProduct['ERP5Security']
dispacher.addERP5DumbHTTPExtractionPlugin(self.test_id)
getattr(pas, self.test_id).manage_activateInterfaces(
('IExtractionPlugin',))
elif len(access_extraction_list) == 1:
self.test_id = access_extraction_list[0].getId()
elif len(access_extraction_list) > 1:
raise ValueError
transaction.commit()
def _createPerson(self, new_id, password=None):
"""Creates a person in person module, and returns the object, after
indexing is done. """
person_module = self.getPersonModule()
person = person_module.newContent(portal_type='Person',
reference='TESTP-' + new_id)
if password:
person.setPassword(password)
person.newContent(portal_type = 'Assignment').open()
transaction.commit()
return person
def test_working_authentication(self):
person = self.person = self._createPerson(self.new_id, "test")
request = self.do_fake_request("GET", {"HTTP_AUTHORIZATION": "Basic " + base64.b64encode("%s:test" % self.new_id)})
ret = ERP5DumbHTTPExtractionPlugin("default_extraction").extractCredentials(request)
self.assertEquals(ret, {'login': self.new_id, 'password': 'test', 'remote_host': 'bobo.remote.host', 'remote_address': '204.183.226.81 '})
bt5/erp5_access_token/PropertySheetTemplateItem/portal_property_sheets/TemplateToolERP5AccessTokenExtractionPluginConstraint.xml → bt5/erp5_access_token/TestTemplateItem/portal_components/test.erp5.testERP5DumbHTTPExtractionPlugin.xml
View file @ 79b5a1ec
...@@ -2,65 +2,122 @@ ...@@ -2,65 +2,122 @@
<ZopeData> <ZopeData>
<record id="1" aka="AAAAAAAAAAE="> <record id="1" aka="AAAAAAAAAAE=">
<pickle> <pickle>
<global name="Property Sheet" module="erp5.portal_type"/> <global name="Test Component" module="erp5.portal_type"/>
</pickle> </pickle>
<pickle> <pickle>
<dictionary> <dictionary>
<item> <item>
<key> <string>_count</string> </key> <key> <string>_recorded_property_dict</string> </key>
<value> <value>
<persistent> <string encoding="base64">AAAAAAAAAAI=</string> </persistent> <persistent> <string encoding="base64">AAAAAAAAAAI=</string> </persistent>
</value> </value>
</item> </item>
<item> <item>
<key> <string>_mt_index</string> </key> <key> <string>default_reference</string> </key>
<value> <string>testERP5DumbHTTPExtractionPlugin</string> </value>
</item>
<item>
<key> <string>description</string> </key>
<value> <value>
<persistent> <string encoding="base64">AAAAAAAAAAM=</string> </persistent> <none/>
</value> </value>
</item> </item>
<item> <item>
<key> <string>_tree</string> </key> <key> <string>id</string> </key>
<value> <string>test.erp5.testERP5DumbHTTPExtractionPlugin</string> </value>
</item>
<item>
<key> <string>portal_type</string> </key>
<value> <string>Test Component</string> </value>
</item>
<item>
<key> <string>sid</string> </key>
<value> <value>
<persistent> <string encoding="base64">AAAAAAAAAAQ=</string> </persistent> <none/>
</value> </value>
</item> </item>
<item> <item>
<key> <string>description</string> </key> <key> <string>text_content_error_message</string> </key>
<value> <value>
<none/> <tuple/>
</value> </value>
</item> </item>
<item> <item>
<key> <string>id</string> </key> <key> <string>text_content_warning_message</string> </key>
<value> <string>TemplateToolERP5AccessTokenExtractionPluginConstraint</string> </value> <value>
<tuple/>
</value>
</item> </item>
<item> <item>
<key> <string>portal_type</string> </key> <key> <string>version</string> </key>
<value> <string>Property Sheet</string> </value> <value> <string>erp5</string> </value>
</item>
<item>
<key> <string>workflow_history</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAM=</string> </persistent>
</value>
</item> </item>
</dictionary> </dictionary>
</pickle> </pickle>
</record> </record>
<record id="2" aka="AAAAAAAAAAI="> <record id="2" aka="AAAAAAAAAAI=">
<pickle> <pickle>
<global name="Length" module="BTrees.Length"/> <global name="PersistentMapping" module="Persistence.mapping"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>data</string> </key>
<value>
<dictionary/>
</value>
</item>
</dictionary>
</pickle> </pickle>
<pickle> <int>0</int> </pickle>
</record> </record>
<record id="3" aka="AAAAAAAAAAM="> <record id="3" aka="AAAAAAAAAAM=">
<pickle> <pickle>
<global name="OOBTree" module="BTrees.OOBTree"/> <global name="PersistentMapping" module="Persistence.mapping"/>
</pickle> </pickle>
<pickle> <pickle>
<none/> <dictionary>
<item>
<key> <string>data</string> </key>
<value>
<dictionary>
<item>
<key> <string>component_validation_workflow</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAQ=</string> </persistent>
</value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</pickle> </pickle>
</record> </record>
<record id="4" aka="AAAAAAAAAAQ="> <record id="4" aka="AAAAAAAAAAQ=">
<pickle> <pickle>
<global name="OOBTree" module="BTrees.OOBTree"/> <global name="WorkflowHistoryList" module="Products.ERP5Type.patches.WorkflowTool"/>
</pickle> </pickle>
<pickle> <pickle>
<none/> <tuple>
<none/>
<list>
<dictionary>
<item>
<key> <string>action</string> </key>
<value> <string>validate</string> </value>
</item>
<item>
<key> <string>validation_state</string> </key>
<value> <string>validated</string> </value>
</item>
</dictionary>
</list>
</tuple>
</pickle> </pickle>
</record> </record>
</ZopeData> </ZopeData>
bt5/erp5_access_token/bt/template_portal_type_property_sheet_list
View file @ 79b5a1ec
One Time Restricted Access Token | Url One Time Restricted Access Token | Url
Restricted Access Token | Reference Restricted Access Token | Reference
Restricted Access Token | Url Restricted Access Token | Url
Template Tool | TemplateToolERP5AccessTokenExtractionPluginConstraint \ No newline at end of file
\ No newline at end of file
bt5/erp5_access_token/bt/template_property_sheet_id_list deleted 100644 → 0
View file @ 0229018e
TemplateToolERP5AccessTokenExtractionPluginConstraint
\ No newline at end of file
bt5/erp5_access_token/bt/template_test_id_list
View file @ 79b5a1ec
test.erp5.testERP5AccessToken test.erp5.testERP5AccessTokenAlarm
\ No newline at end of file test.erp5.testERP5AccessTokenSkins
test.erp5.testERP5DumbHTTPExtractionPlugin
\ No newline at end of file
product/ERP5Security/ERP5AccessTokenExtractionPlugin.py
View file @ 79b5a1ec
...@@ -59,49 +59,35 @@ class ERP5AccessTokenExtractionPlugin(BasePlugin): ...@@ -59,49 +59,35 @@ class ERP5AccessTokenExtractionPlugin(BasePlugin):
#ILoginPasswordHostExtractionPlugin# #ILoginPasswordHostExtractionPlugin#
#################################### ####################################
security.declarePrivate('extractCredentials') security.declarePrivate('extractCredentials')
@UnrestrictedMethod
def extractCredentials(self, request): def extractCredentials(self, request):
""" Extract credentials from the request header. """ """ Extract CookieHash credentials from the request header. """
creds = {} creds = {}
# Extract token from HTTP Header # XXX Extract from HTTP Header, URL parameter are hardcoded.
token = request.getHeader("X-ACCESS-TOKEN", request.form.get("access_token", None)) # More flexible way would be to configure on the portal type level
if token: token = request.getHeader("X-ACCESS-TOKEN", None)
creds['erp5_access_token_id'] = token if token is None:
creds['remote_host'] = request.get('REMOTE_HOST', '') token = request.form.get("access_token", None)
try: if token is not None:
creds['remote_address'] = request.getClientAddr()
except AttributeError:
creds['remote_address'] = request.get('REMOTE_ADDR', '')
return creds
#######################
#IAuthenticationPlugin#
#######################
security.declarePrivate('authenticateCredentials')
@UnrestrictedMethod
def authenticateCredentials(self, credentials):
""" Map credentials to a user ID. """
if 'erp5_access_token_id' in credentials:
erp5_access_token_id = credentials['erp5_access_token_id']
token_document = self.getPortalObject().access_token_module.\ token_document = self.getPortalObject().access_token_module.\
_getOb(erp5_access_token_id, None) _getOb(token, None)
# Access Token should be validated # Access Token should be validated
# Check restricted access of URL # Check restricted access of URL
# Extract login information # Extract login information
if token_document is not None: if token_document is not None:
# Token API changed from returning a login to returning a user id. external_login = None
# We detect if the old way of configuration is still in place and
# advise that configuration has to be updated in that case.
method = token_document._getTypeBasedMethod('getExternalLogin') method = token_document._getTypeBasedMethod('getExternalLogin')
assert method is None, "Please update and remove obsolete method %r" % method
user_id = None
method = token_document._getTypeBasedMethod('getUserId')
if method is not None: if method is not None:
user_id = method() external_login = method()
if user_id is not None: if external_login is not None:
return (user_id, 'token {erp5_access_token_id} for {user_id}'.format(**locals())) creds['external_login'] = external_login
creds['remote_host'] = request.get('REMOTE_HOST', '')
try:
creds['remote_address'] = request.getClientAddr()
except AttributeError:
creds['remote_address'] = request.get('REMOTE_ADDR', '')
return creds
#Form for new plugin in ZMI #Form for new plugin in ZMI
manage_addERP5AccessTokenExtractionPluginForm = PageTemplateFile( manage_addERP5AccessTokenExtractionPluginForm = PageTemplateFile(
...@@ -123,7 +109,6 @@ def addERP5AccessTokenExtractionPlugin(dispatcher, id, title=None, REQUEST=None) ...@@ -123,7 +109,6 @@ def addERP5AccessTokenExtractionPlugin(dispatcher, id, title=None, REQUEST=None)
#List implementation of class #List implementation of class
classImplements(ERP5AccessTokenExtractionPlugin, classImplements(ERP5AccessTokenExtractionPlugin,
plugins.ILoginPasswordHostExtractionPlugin, plugins.ILoginPasswordHostExtractionPlugin
plugins.IAuthenticationPlugin, )
)
InitializeClass(ERP5AccessTokenExtractionPlugin) InitializeClass(ERP5AccessTokenExtractionPlugin)
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7